from __future__ import annotations import asyncio import os import tempfile import unittest from unittest.mock import patch from fastapi.testclient import TestClient from fastapi import HTTPException import backend.main as main class ApiRegressionTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.client = TestClient(main.app) cls.symbol_a, cls.symbol_b = list(main.SYMBOLS.keys())[:2] def _build_current_forecast_payload( self, requested: dict[str, bool] | None = None, ) -> dict[str, object]: requested_models = dict(main.DEFAULT_FORECAST_MODEL_SELECTION) if requested: requested_models.update(requested) if not any(requested_models.values()): requested_models["kronos"] = True def build_model_meta(model_key: str) -> dict[str, object]: adapter_mode = { "kronos": "kronos_ohlc4_proxy", "timesfm": "timesfm_native", "chronos": "chronos_native", }[model_key] is_kronos = model_key == "kronos" return { "model_key": model_key, "name": model_key, "context_length": 64, "available_history": 500, "input_semantics": { "feature_channels": ["ohlc4"], "price_mode": "ohlc4_single_channel", "base_signal": "ohlc4", "volume_mode": "synthetic_zero" if is_kronos else "omitted", "amount_mode": "synthetic_zero" if is_kronos else "omitted", "active_forecast_channels": ["ohlc4"], "adapter_mode": adapter_mode, }, "output_semantics": { "forecast_channel": "ohlc4", "forecast_mode": "single_future_ohlc4_line", "candle_projection": "omitted", "reference_baseline": "last_ohlc4", }, } forecast_rows = [ {"time": 1, "p10": 1.0, "p50": 1.1, "p90": 1.2}, {"time": 2, "p10": 1.1, "p50": 1.2, "p90": 1.3}, ] forecast_models: dict[str, object] = {} components: dict[str, object] = {} active_models: list[str] = [] for model_key in main.FORECAST_MODEL_ORDER: enabled = bool(requested_models.get(model_key, False)) if enabled: model_meta = build_model_meta(model_key) forecast_models[model_key] = { "enabled": True, "available": True, "success": True, "skipped": False, "error": None, "forecast": forecast_rows, "model": model_meta, "ensemble": {"confidence": 61.0}, "model_diagnostics": {}, } components[model_key] = model_meta active_models.append(model_key) else: forecast_models[model_key] = { "enabled": False, "available": True, "success": False, "skipped": True, "error": None, "forecast": [], "model": {}, "ensemble": {}, "model_diagnostics": {}, } return { "forecast": forecast_rows, "display": { "mode": ( "raw_kronos_ohlc4_line" if active_models == ["kronos"] else "raw_timesfm_ohlc4_line" if active_models == ["timesfm"] else "raw_chronos_ohlc4_line" if active_models == ["chronos"] else "multi_model_ohlc4_line" ), "channels": ["ohlc4"], "output_mode": "single_future_ohlc4_line", "reference_series": "ohlc4", "market_price_field": "last_close", "forecast_reference_field": "last_ohlc4", "visual_anchor_field": "last_close", "combination_mode": "mean_of_enabled_models", }, "forecast_models": forecast_models, "model_selection": { "requested": requested_models, "active": active_models, "defaults": dict(main.DEFAULT_FORECAST_MODEL_SELECTION), "combination_mode": "mean_of_enabled_models", }, "model": { "name": active_models[0] if len(active_models) == 1 else "mean_of_enabled_models", "active_models": active_models, "components": components, "cache_version": main.CACHE_VERSION, }, } def test_cache_stats_requires_runtime_admin_token(self) -> None: with patch.object(main, "ADMIN_TOKEN", "test-admin-token"): unauthorized = self.client.get("/api/cache/stats") self.assertEqual(unauthorized.status_code, 401) wrong_token = self.client.get( "/api/cache/stats", headers={"X-Admin-Token": "wrong-token"}, ) self.assertEqual(wrong_token.status_code, 401) authorized = self.client.get( "/api/cache/stats", headers={"X-Admin-Token": "test-admin-token"}, ) self.assertEqual(authorized.status_code, 200) def test_forecast_rules_endpoint_exposes_versioned_contract(self) -> None: response = self.client.get("/api/forecasting/rules") self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["rules"]["version"], main.FORECAST_RULE_DOCUMENT.version) self.assertEqual(body["rules"]["default_horizon"], 10) self.assertEqual(body["rules"]["recommended_context_length"], 512) self.assertEqual(body["rules"]["default_line_width"], 1) self.assertIn("chronos", body["models"]) def test_frontend_forecast_model_registry_has_dedicated_no_cache_route(self) -> None: response = self.client.get("/forecast-models.js") self.assertEqual(response.status_code, 200) self.assertIn("Cache-Control", response.headers) self.assertEqual( response.headers["Cache-Control"], "no-store, no-cache, must-revalidate, max-age=0", ) def test_forecast_model_selection_supports_chronos(self) -> None: selection = main._normalize_forecast_model_selection( use_kronos=False, use_timesfm=False, use_chronos=True, ) self.assertEqual( selection, {"kronos": False, "timesfm": False, "chronos": True}, ) def test_watchlist_tickers_deduplicates_and_reports_invalid_symbols(self) -> None: async def fake_fetch_ticker(symbol: str) -> dict[str, object]: return {"symbol": symbol, "price": 123.45} payload = { "symbols": [ self.symbol_a, self.symbol_a.lower(), "BINANCE:" + self.symbol_b, "NOT_A_REAL_SYMBOL", ] } with patch.object(main, "fetch_ticker", side_effect=fake_fetch_ticker): response = self.client.post("/api/watchlist/tickers", json=payload) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["count"], 2) self.assertEqual(body["duplicate_count"], 1) self.assertEqual(body["invalid_symbols"], ["NOT_A_REAL_SYMBOL"]) self.assertEqual(sorted(body["tickers"].keys()), sorted([self.symbol_a, self.symbol_b])) def test_backtest_rejects_invalid_interval_before_fetch(self) -> None: response = self.client.get(f"/api/backtest/{self.symbol_a}", params={"interval": "13h"}) self.assertEqual(response.status_code, 400) self.assertIn("Unsupported interval", response.json()["detail"]) def test_volume_profile_rejects_invalid_bucket_count(self) -> None: response = self.client.get( f"/api/volume-profile/{self.symbol_a}", params={"buckets": 0}, ) self.assertEqual(response.status_code, 422) def test_ping_exposes_runtime_version(self) -> None: response = self.client.get("/api/ping") self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {"status": "ok", "version": main.APP_VERSION}) self.assertIn("X-Request-ID", response.headers) self.assertIn("X-Response-Time-Ms", response.headers) def test_real_strength_catalog_groups_dxy_usdx_and_strength_indexes(self) -> None: response = self.client.get("/api/symbols") self.assertEqual(response.status_code, 200) body = response.json() self.assertIn("Real Strength", body["categories"]) symbols = {entry["symbol"]: entry for entry in body["symbols"]} for symbol in ["DXY", "USDX", "EURX", "GBPX", "CHFX", "JPYX", "CADX", "AUDX", "NZDX"]: self.assertIn(symbol, symbols) self.assertEqual(symbols[symbol]["category"], "Real Strength") for symbol in ["USDX", "EURX", "GBPX", "CHFX", "JPYX", "CADX", "AUDX", "NZDX"]: self.assertEqual(symbols[symbol]["sources"], ["synthetic"]) self.assertEqual(symbols["DXY"]["sources"], ["twelvedata", "yfinance"]) def test_forex_source_priority_excludes_binance(self) -> None: self.assertEqual(main._get_source_priority("EURUSD"), ["twelvedata", "yfinance"]) self.assertNotIn("binance", main._get_source_priority("EURUSD")) self.assertEqual(main._get_source_priority("DXY"), ["yfinance", "twelvedata"]) def test_xauusd_priority_prefers_binance_paxg_proxy_first(self) -> None: self.assertEqual( main._get_source_priority("XAUUSD", "1d"), ["binance", "twelvedata", "yfinance"], ) def test_forecast_context_window_prefers_recent_tail(self) -> None: self.assertEqual(main._resolve_forecast_context_window("1h", 1000), 256) self.assertEqual(main._resolve_forecast_context_window("1d", 180), 180) self.assertEqual(main._resolve_forecast_context_window("1w", 400), 192) def test_forecast_context_candidates_include_shorter_windows_for_responsiveness(self) -> None: self.assertEqual(main._resolve_forecast_context_candidates("1d", 400), [48, 64, 128, 256]) self.assertEqual(main._resolve_forecast_context_candidates("4h", 300), [64, 96, 128, 256]) self.assertEqual(main._resolve_forecast_context_candidates("1d", 60), [48, 60]) def test_kronos_context_candidates_cap_direct_model_window_for_latency(self) -> None: self.assertEqual(main._resolve_kronos_context_candidates("1d", 1000), [64, 128, 256]) self.assertEqual(main._resolve_model_context_cap("kronos", "1d", 1000), 256) def test_forecast_context_candidate_score_penalizes_near_flat_paths(self) -> None: responsive = main._score_forecast_context_candidate( analysis_bundle={ "p50": main.np.array([100.0, 99.4, 98.8, 98.1], dtype=float), "confidence": 58.0, }, last_ohlc4=100.0, recent_abs_step_pct=1.0, ) flat = main._score_forecast_context_candidate( analysis_bundle={ "p50": main.np.array([100.0, 99.98, 100.01, 100.0], dtype=float), "confidence": 58.0, }, last_ohlc4=100.0, recent_abs_step_pct=1.0, ) self.assertGreater(responsive["score"], flat["score"]) self.assertGreater(responsive["step_abs_mean_pct"], flat["step_abs_mean_pct"]) def test_forecast_amplitude_calibration_scales_flat_path_toward_historical_targets(self) -> None: raw_bundle = main._build_raw_ohlc4_bundle( { "p10": main.np.array([99.98, 99.96, 99.94], dtype=float), "p50": main.np.array([100.0, 99.99, 99.98], dtype=float), "p90": main.np.array([100.02, 100.01, 100.0], dtype=float), }, 100.0, ) calibrated_bundle, calibration_meta = main._apply_forecast_amplitude_calibration( raw_bundle=raw_bundle, last_ohlc4=100.0, target_profile={ "target_step_abs_mean_pct": 0.45, "target_range_pct": 1.2, "recent_abs_step_pct": 0.8, "rolling_targets": None, "regime_targets": None, }, ) self.assertGreater(calibration_meta["scale"], 1.0) self.assertGreater( calibrated_bundle["path_metrics"]["final_return_pct"], raw_bundle["path_metrics"]["final_return_pct"] * 3, ) self.assertGreater( calibration_meta["calibrated_path"]["step_abs_mean_pct"], calibration_meta["raw_path"]["step_abs_mean_pct"], ) def test_forecast_path_texture_adds_stepwise_zigzag_to_future_points(self) -> None: base_bundle = main._build_raw_ohlc4_bundle( { "p10": main.np.array([99.6, 99.4, 99.2, 99.0], dtype=float), "p50": main.np.array([99.9, 99.8, 99.7, 99.6], dtype=float), "p90": main.np.array([100.2, 100.1, 100.0, 99.9], dtype=float), }, 100.0, ) textured_bundle, texture_meta = main._apply_forecast_path_texture( base_bundle=base_bundle, last_ohlc4=100.0, target_profile={ "target_step_abs_mean_pct": 0.45, "target_range_pct": 1.4, "recent_abs_step_pct": 0.8, }, texture_template={ "step_returns_pct": main.np.array([0.22, -0.35, 0.28, -0.12], dtype=float), "step_abs_mean_pct": 0.3133, "range_pct": 0.40, "match_count": 6, "signature_len": 48, }, ) self.assertTrue(texture_meta["applied"]) self.assertGreater(texture_meta["blend_alpha"], 0.0) self.assertGreater( texture_meta["textured_path"]["step_abs_mean_pct"], texture_meta["base_path"]["step_abs_mean_pct"], ) self.assertNotAlmostEqual( textured_bundle["path_metrics"]["final_return_pct"], base_bundle["path_metrics"]["final_return_pct"], delta=0.02, ) def test_crypto_dynamic_source_mappings_are_derived_for_hf_safe_fallbacks(self) -> None: self.assertEqual(main._get_symbol_mapping("BTCUSD", "twelvedata"), "BTC/USD") self.assertEqual(main._get_symbol_mapping("BTCUSD", "yfinance"), "BTC-USD") self.assertEqual(main._get_symbol_mapping("BTCUSD", "finnhub"), "BINANCE:BTCUSDT") self.assertEqual(main._get_canonical_symbol("BTC-USD"), "BTCUSD") def test_crypto_source_priority_prefers_hf_safe_providers_on_hf(self) -> None: with patch.object(main.settings, "is_hf", True): self.assertEqual( main._get_source_priority("BTCUSD", "1d"), ["yfinance", "twelvedata", "finnhub", "coingecko", "binance", "bybit"], ) self.assertEqual( main._get_source_priority("BTCUSD", "1h"), ["twelvedata", "yfinance", "finnhub", "coingecko", "binance", "bybit"], ) def test_crypto_historical_fetch_on_hf_uses_hf_safe_fallback_first(self) -> None: sample_rows = [ {"time": i, "open": 1.0, "high": 1.1, "low": 0.9, "close": 1.0, "volume": 0.0} for i in range(1, 41) ] attempts: list[str] = [] async def fake_twelvedata(symbol: str, interval: str, limit: int) -> list[dict[str, float]]: attempts.append("twelvedata") return sample_rows[-limit:] async def fake_yfinance(symbol: str, interval: str, limit: int) -> list[dict[str, float]]: attempts.append("yfinance") raise RuntimeError("should not be needed after TwelveData success") with patch.object(main.settings, "is_hf", True), patch.object( main, "fetch_twelvedata", side_effect=fake_twelvedata, ), patch.object( main, "fetch_yfinance", side_effect=fake_yfinance, ): rows, source = asyncio.run( main._run_historical_fetch("BTCUSD", "1h", 40, "hist_btcusd_1h_hf_test") ) self.assertEqual(source, "twelvedata") self.assertEqual(len(rows), 40) self.assertEqual(attempts, ["twelvedata"]) def test_historical_fetch_falls_back_after_provider_http_error(self) -> None: sample_rows = [ {"time": i, "open": 1.0, "high": 1.1, "low": 0.9, "close": 1.0, "volume": 0.0} for i in range(1, 41) ] async def fake_twelvedata(symbol: str, interval: str, limit: int) -> list[dict[str, float]]: raise HTTPException(status_code=429, detail="rate limit") async def fake_yfinance(symbol: str, interval: str, limit: int) -> list[dict[str, float]]: return sample_rows[-limit:] with patch.object(main, "fetch_twelvedata", side_effect=fake_twelvedata), patch.object( main, "fetch_yfinance", side_effect=fake_yfinance, ): rows, source = asyncio.run( main._run_historical_fetch("EURUSD", "1d", 40, "hist_eurusd_1d_test") ) self.assertEqual(source, "yfinance") self.assertEqual(len(rows), 40) def test_clear_all_cache_rejects_unknown_target(self) -> None: with patch.object(main, "ADMIN_TOKEN", "test-admin-token"): response = self.client.delete( "/api/cache", params={"target": "invalid-target"}, headers={"X-Admin-Token": "test-admin-token"}, ) self.assertEqual(response.status_code, 400) self.assertIn("Unsupported cache target", response.json()["detail"]) def test_clear_all_cache_accepts_indicators_target(self) -> None: with patch.object(main, "ADMIN_TOKEN", "test-admin-token"): response = self.client.delete( "/api/cache", params={"target": "indicators"}, headers={"X-Admin-Token": "test-admin-token"}, ) self.assertEqual(response.status_code, 200) self.assertIn("indicators", response.json()["cleared"]) def test_metrics_include_request_observability_snapshot(self) -> None: self.client.get("/api/ping") with patch.object(main, "ADMIN_TOKEN", "test-admin-token"): response = self.client.get( "/api/metrics", headers={"X-Admin-Token": "test-admin-token"}, ) self.assertEqual(response.status_code, 200) body = response.json() self.assertIn("request_metrics", body) self.assertIn("/api/ping", body["request_metrics"]["routes"]) def test_vote_gauge_uses_equal_weight_formula(self) -> None: gauge = main._calc_vote_gauge(buy=4, sell=12, neutral=10) self.assertAlmostEqual(gauge, 34.6153846154, places=6) def test_technical_score_uses_total_votes_without_bonus_bias(self) -> None: tech = main._calc_technical_score_v2( osc_score={ "gauge": 80.0, "buy": 2, "sell": 6, "neutral": 4, "buy_weight": 10.0, "sell_weight": 10.0, "neutral_weight": 10.0, }, ma_score={ "gauge": 20.0, "buy": 2, "sell": 6, "neutral": 6, "buy_weight": 10.0, "sell_weight": 10.0, "neutral_weight": 10.0, "golden_cross": True, "death_cross": False, }, regime="stable_bull", interval="1h", ) self.assertEqual(tech["buy"], 4) self.assertEqual(tech["sell"], 12) self.assertEqual(tech["neutral"], 10) self.assertAlmostEqual(tech["gauge"], 34.6, places=1) def test_ma_score_cross_flags_are_plain_python_bools(self) -> None: ma_score = main._calc_ma_score( ma_data=[ {"name": "EMA1 / EMA5", "key": "ema_1_5", "action": "Mua"}, {"name": "EMA5 / EMA10", "key": "ema_5_10", "action": "Mua"}, ], closes=main.np.linspace(1.0, 10.0, 220), interval="1d", ) self.assertIs(type(ma_score["golden_cross"]), bool) self.assertIs(type(ma_score["death_cross"]), bool) def test_local_verdict_is_fast_and_deterministic(self) -> None: verdict = main._derive_local_verdict( analysis={ "summary": {"bias": "bullish"}, "technicals": {"gauge": 63.0}, "ai_gauge": {"gauge": 61.0, "confidence_pct": 58.0}, "multi_timeframe": {"alignment": True}, }, forecast_pct=1.2, ) self.assertEqual(verdict, "Mua ngay") def test_summary_score_uses_ai_certainty_weighting(self) -> None: summary = main._calc_summary_score_v2( tech_score={ "gauge": 30.0, "buy": 4, "sell": 12, "neutral": 10, "buy_weight": 0.2, "sell_weight": 0.5, "neutral_weight": 0.3, "components": { "oscillators": 28.0, "moving_averages": 32.0, }, }, ai_score={ "gauge": 35.0, "certainty": 40.0, }, interval="1d", ) self.assertEqual(summary["gauge"], 32.0) self.assertEqual(summary["components"]["ai_weight"], 0.4) self.assertEqual(summary["components"]["technical_weight"], 0.6) def test_synthetic_component_candles_support_product_and_ratio(self) -> None: product = main._combine_component_candles( main.SyntheticComponentSpec(name="EURCAD", mode="product", left_symbol="EURUSD", right_symbol="USDCAD"), {"open": 1.1, "high": 1.2, "low": 1.0, "close": 1.15}, {"open": 1.3, "high": 1.4, "low": 1.2, "close": 1.35}, ) ratio = main._combine_component_candles( main.SyntheticComponentSpec(name="EURAUD", mode="ratio", left_symbol="EURUSD", right_symbol="AUDUSD"), {"open": 1.1, "high": 1.2, "low": 1.0, "close": 1.15}, {"open": 0.7, "high": 0.8, "low": 0.6, "close": 0.75}, ) inverse = main._combine_component_candles( main.SyntheticComponentSpec(name="EURGBP", mode="inverse", left_symbol="EURGBP"), {"open": 0.85, "high": 0.86, "low": 0.84, "close": 0.855}, ) self.assertAlmostEqual(product["open"], 1.43, places=6) self.assertAlmostEqual(product["high"], 1.68, places=6) self.assertAlmostEqual(product["low"], 1.2, places=6) self.assertAlmostEqual(ratio["open"], 1.5714285714, places=6) self.assertAlmostEqual(ratio["high"], 2.0, places=6) self.assertAlmostEqual(ratio["low"], 1.25, places=6) self.assertAlmostEqual(inverse["open"], 1 / 0.85, places=6) self.assertAlmostEqual(inverse["high"], 1 / 0.84, places=6) self.assertAlmostEqual(inverse["low"], 1 / 0.86, places=6) def test_synthetic_component_candles_use_body_only_extrema_for_intraday_product_and_ratio(self) -> None: direct = main._combine_component_candles( main.SyntheticComponentSpec(name="EURUSD", mode="direct", left_symbol="EURUSD"), {"open": 1.1, "high": 1.2, "low": 1.0, "close": 1.15}, use_body_only_extrema=True, ) inverse = main._combine_component_candles( main.SyntheticComponentSpec(name="EURGBP", mode="inverse", left_symbol="EURGBP"), {"open": 0.85, "high": 0.86, "low": 0.84, "close": 0.855}, use_body_only_extrema=True, ) product = main._combine_component_candles( main.SyntheticComponentSpec(name="EURCAD", mode="product", left_symbol="EURUSD", right_symbol="USDCAD"), {"open": 1.1, "high": 1.2, "low": 1.0, "close": 1.15}, {"open": 1.3, "high": 1.4, "low": 1.2, "close": 1.35}, use_body_only_extrema=True, ) ratio = main._combine_component_candles( main.SyntheticComponentSpec(name="EURAUD", mode="ratio", left_symbol="EURUSD", right_symbol="AUDUSD"), {"open": 1.1, "high": 1.2, "low": 1.0, "close": 1.15}, {"open": 0.7, "high": 0.8, "low": 0.6, "close": 0.75}, use_body_only_extrema=True, ) self.assertAlmostEqual(direct["open"], 1.1, places=6) self.assertAlmostEqual(direct["close"], 1.15, places=6) self.assertAlmostEqual(direct["high"], 1.15, places=6) self.assertAlmostEqual(direct["low"], 1.1, places=6) self.assertAlmostEqual(inverse["open"], 1 / 0.85, places=6) self.assertAlmostEqual(inverse["close"], 1 / 0.855, places=6) self.assertAlmostEqual(inverse["high"], 1 / 0.85, places=6) self.assertAlmostEqual(inverse["low"], 1 / 0.855, places=6) self.assertAlmostEqual(product["open"], 1.43, places=6) self.assertAlmostEqual(product["close"], 1.5525, places=6) self.assertAlmostEqual(product["high"], 1.5525, places=6) self.assertAlmostEqual(product["low"], 1.43, places=6) self.assertAlmostEqual(ratio["open"], 1.5714285714, places=6) self.assertAlmostEqual(ratio["close"], 1.5333333333, places=6) self.assertAlmostEqual(ratio["high"], 1.5714285714, places=6) self.assertAlmostEqual(ratio["low"], 1.5333333333, places=6) def test_synthetic_eurx_history_builds_from_component_series(self) -> None: base_rows = { "EURUSD": [ {"time": 1, "open": 1.10, "high": 1.11, "low": 1.09, "close": 1.105, "volume": 100.0}, {"time": 2, "open": 1.11, "high": 1.12, "low": 1.10, "close": 1.115, "volume": 100.0}, {"time": 3, "open": 1.12, "high": 1.13, "low": 1.11, "close": 1.125, "volume": 100.0}, ], "EURGBP": [ {"time": 1, "open": 0.85, "high": 0.86, "low": 0.84, "close": 0.855, "volume": 100.0}, {"time": 2, "open": 0.855, "high": 0.865, "low": 0.845, "close": 0.86, "volume": 100.0}, {"time": 3, "open": 0.86, "high": 0.87, "low": 0.85, "close": 0.865, "volume": 100.0}, ], "USDCHF": [ {"time": 1, "open": 0.90, "high": 0.91, "low": 0.89, "close": 0.905, "volume": 100.0}, {"time": 2, "open": 0.905, "high": 0.915, "low": 0.895, "close": 0.91, "volume": 100.0}, {"time": 3, "open": 0.91, "high": 0.92, "low": 0.90, "close": 0.915, "volume": 100.0}, ], "EURJPY": [ {"time": 1, "open": 160.0, "high": 161.0, "low": 159.0, "close": 160.5, "volume": 100.0}, {"time": 2, "open": 160.5, "high": 161.5, "low": 159.5, "close": 161.0, "volume": 100.0}, {"time": 3, "open": 161.0, "high": 162.0, "low": 160.0, "close": 161.5, "volume": 100.0}, ], "USDCAD": [ {"time": 1, "open": 1.35, "high": 1.36, "low": 1.34, "close": 1.355, "volume": 100.0}, {"time": 2, "open": 1.355, "high": 1.365, "low": 1.345, "close": 1.36, "volume": 100.0}, {"time": 3, "open": 1.36, "high": 1.37, "low": 1.35, "close": 1.365, "volume": 100.0}, ], "AUDUSD": [ {"time": 1, "open": 0.66, "high": 0.67, "low": 0.65, "close": 0.665, "volume": 100.0}, {"time": 2, "open": 0.665, "high": 0.675, "low": 0.655, "close": 0.67, "volume": 100.0}, {"time": 3, "open": 0.67, "high": 0.68, "low": 0.66, "close": 0.675, "volume": 100.0}, ], "NZDUSD": [ {"time": 1, "open": 0.61, "high": 0.62, "low": 0.60, "close": 0.615, "volume": 100.0}, {"time": 2, "open": 0.615, "high": 0.625, "low": 0.605, "close": 0.62, "volume": 100.0}, {"time": 3, "open": 0.62, "high": 0.63, "low": 0.61, "close": 0.625, "volume": 100.0}, ], } async def fake_fetch_from_source( source: str, symbol: str, interval: str, limit: int, ) -> tuple[list[dict[str, float]], str]: self.assertIn(source, {"twelvedata", "finnhub", "yfinance"}) return base_rows[symbol][-limit:] with patch.object(main, "_fetch_historical_from_source", side_effect=fake_fetch_from_source): rows, source = asyncio.run( main._build_synthetic_symbol_history("EURX", "1h", 3, "test-cache-key") ) self.assertEqual(len(rows), 3) self.assertEqual(rows[-1]["time"], 3) self.assertEqual(rows[-1]["volume"], 0.0) self.assertEqual(source, "synthetic:twelvedata") self.assertGreater(rows[-1]["close"], 0.0) def test_synthetic_ticker_uses_requested_interval_instead_of_forcing_5m(self) -> None: calls: list[tuple[str, str, int, int]] = [] async def fake_fetch_historical( symbol: str, interval: str, limit: int, min_context: int = 0, **_: object, ) -> tuple[list[dict[str, float]], str]: calls.append((symbol, interval, limit, min_context)) rows = [ {"time": 1, "open": 1.0, "high": 1.1, "low": 0.9, "close": 1.0, "volume": 0.0}, {"time": 2, "open": 1.0, "high": 1.2, "low": 0.95, "close": 1.1, "volume": 0.0}, ] return rows, "synthetic:test" with patch.object(main, "fetch_historical", side_effect=fake_fetch_historical): ticker = asyncio.run(main.fetch_ticker("CHFX", interval="1d")) self.assertEqual(ticker["symbol"], "CHFX") self.assertEqual(calls, [("CHFX", "1d", 2, 2)]) def test_synthetic_history_uses_single_source_and_requested_interval(self) -> None: calls: list[tuple[str, str, str, int]] = [] async def fake_fetch_from_source( source: str, symbol: str, interval: str, limit: int, ) -> list[dict[str, float]]: calls.append((source, symbol, interval, limit)) return [ {"time": 1, "open": 1.0, "high": 1.1, "low": 0.9, "close": 1.0, "volume": 0.0}, {"time": 2, "open": 1.0, "high": 1.2, "low": 0.95, "close": 1.1, "volume": 0.0}, {"time": 3, "open": 1.1, "high": 1.25, "low": 1.0, "close": 1.15, "volume": 0.0}, ] with patch.object(main, "_fetch_historical_from_source", side_effect=fake_fetch_from_source): rows, source = asyncio.run( main._build_synthetic_symbol_history("CHFX", "4h", 3, "synthetic-chfx-4h") ) self.assertEqual(len(rows), 3) self.assertEqual(source, "synthetic:twelvedata") self.assertTrue(calls) self.assertEqual({item[0] for item in calls}, {"twelvedata"}) self.assertEqual({item[2] for item in calls}, {"4h"}) def test_timesfm_feature_prep_collapses_prices_to_ohlc4(self) -> None: df = main.pd.DataFrame( [ {"open": 1.0, "high": 1.1, "low": 0.9, "close": 1.05, "volume": 0.0, "amount": 0.0}, {"open": 1.05, "high": 1.15, "low": 0.95, "close": 1.1, "volume": 0.0, "amount": 0.0}, ] ) prepared = main.TimesFMForecaster._prepare_feature_frame(df) expected_ohlc4 = df[["open", "high", "low", "close"]].mean(axis=1).astype(main.np.float32) self.assertEqual(list(prepared.columns), ["ohlc4"]) self.assertTrue(main.np.allclose(prepared["ohlc4"].values, expected_ohlc4.values)) def test_timesfm_feature_prep_ignores_upstream_volume_and_amount_noise(self) -> None: df = main.pd.DataFrame( [ {"open": 10.0, "high": 11.0, "low": 9.0, "close": 10.5, "volume": 100.0, "amount": 1050.0}, {"open": 11.0, "high": 12.0, "low": 10.0, "close": 11.5, "volume": 200.0, "amount": 2300.0}, ] ) prepared = main.TimesFMForecaster._prepare_feature_frame(df) expected_ohlc4 = df[["open", "high", "low", "close"]].mean(axis=1).astype(main.np.float32) self.assertEqual(list(prepared.columns), ["ohlc4"]) self.assertTrue(main.np.allclose(prepared["ohlc4"].values, expected_ohlc4.values)) def test_timesfm_input_series_uses_float32_ohlc4_contract(self) -> None: df = main.pd.DataFrame( [ {"open": 1.0, "high": 1.1, "low": 0.9, "close": 1.0, "volume": 0.0}, {"open": 1.1, "high": 1.2, "low": 1.0, "close": 1.1, "volume": 0.0}, {"open": 1.2, "high": 1.3, "low": 1.1, "close": 1.2, "volume": 0.0}, {"open": 1.3, "high": 1.4, "low": 1.2, "close": 1.3, "volume": 0.0}, {"open": 1.4, "high": 1.5, "low": 1.3, "close": 1.4, "volume": 0.0}, {"open": 1.5, "high": 1.6, "low": 1.4, "close": 1.5, "volume": 0.0}, {"open": 1.6, "high": 1.7, "low": 1.5, "close": 1.6, "volume": 0.0}, {"open": 1.7, "high": 1.8, "low": 1.6, "close": 1.7, "volume": 0.0}, {"open": 1.8, "high": 1.9, "low": 1.7, "close": 1.8, "volume": 0.0}, {"open": 1.9, "high": 2.0, "low": 1.8, "close": 1.9, "volume": 0.0}, {"open": 2.0, "high": 2.1, "low": 1.9, "close": 2.0, "volume": 0.0}, {"open": 2.1, "high": 2.2, "low": 2.0, "close": 2.1, "volume": 0.0}, {"open": 2.2, "high": 2.3, "low": 2.1, "close": 2.2, "volume": 0.0}, {"open": 2.3, "high": 2.4, "low": 2.2, "close": 2.3, "volume": 0.0}, {"open": 2.4, "high": 2.5, "low": 2.3, "close": 2.4, "volume": 0.0}, {"open": 2.5, "high": 2.6, "low": 2.4, "close": 2.5, "volume": 0.0}, {"open": 2.6, "high": 2.7, "low": 2.5, "close": 2.6, "volume": 0.0}, {"open": 2.7, "high": 2.8, "low": 2.6, "close": 2.7, "volume": 0.0}, {"open": 2.8, "high": 2.9, "low": 2.7, "close": 2.8, "volume": 0.0}, {"open": 2.9, "high": 3.0, "low": 2.8, "close": 2.9, "volume": 0.0}, {"open": 3.0, "high": 3.1, "low": 2.9, "close": 3.0, "volume": 0.0}, {"open": 3.1, "high": 3.2, "low": 3.0, "close": 3.1, "volume": 0.0}, {"open": 3.2, "high": 3.3, "low": 3.1, "close": 3.2, "volume": 0.0}, {"open": 3.3, "high": 3.4, "low": 3.2, "close": 3.3, "volume": 0.0}, {"open": 3.4, "high": 3.5, "low": 3.3, "close": 3.4, "volume": 0.0}, {"open": 3.5, "high": 3.6, "low": 3.4, "close": 3.5, "volume": 0.0}, {"open": 3.6, "high": 3.7, "low": 3.5, "close": 3.6, "volume": 0.0}, {"open": 3.7, "high": 3.8, "low": 3.6, "close": 3.7, "volume": 0.0}, {"open": 3.8, "high": 3.9, "low": 3.7, "close": 3.8, "volume": 0.0}, {"open": 3.9, "high": 4.0, "low": 3.8, "close": 3.9, "volume": 0.0}, {"open": 4.0, "high": 4.1, "low": 3.9, "close": 4.0, "volume": 0.0}, {"open": 4.1, "high": 4.2, "low": 4.0, "close": 4.1, "volume": 0.0}, {"open": 4.2, "high": 4.3, "low": 4.1, "close": 4.2, "volume": 0.0}, {"open": 4.3, "high": 4.4, "low": 4.2, "close": 4.3, "volume": 0.0}, ] ) ohlc4_series = main.TimesFMForecaster._extract_ohlc4_series(df) expected_ohlc4 = df[["open", "high", "low", "close"]].mean(axis=1).astype(main.np.float32) self.assertEqual(ohlc4_series.dtype, main.np.float32) self.assertTrue(main.np.allclose(ohlc4_series, expected_ohlc4.values)) def test_timesfm_output_validation_prefers_point_forecast_median(self) -> None: point = main.np.array([[101.0, 102.0]], dtype=main.np.float32) quantiles = main.np.array( [ [ [100.5, 99.0, 99.5, 100.0, 100.5, 101.0, 101.5, 102.0, 102.5, 103.0], [101.5, 100.0, 100.5, 101.0, 101.5, 102.0, 102.5, 103.0, 103.5, 104.0], ] ], dtype=main.np.float32, ) p10, p50, p90, diagnostics = main.TimesFMForecaster._validate_output_tensors( point_forecast=point, quantile_forecast=quantiles, horizon=2, ) self.assertTrue(main.np.allclose(p10, [99.0, 100.0])) self.assertTrue(main.np.allclose(p50, point[0])) self.assertTrue(main.np.allclose(p90, [103.0, 104.0])) self.assertEqual(diagnostics["point_shape"], [1, 2]) self.assertEqual(diagnostics["quantile_shape"], [1, 2, 10]) self.assertTrue(diagnostics["median_matches_point_forecast"]) self.assertTrue(diagnostics["quantiles_monotonic"]) def test_forecast_payload_schema_guard_rejects_legacy_blended_payload(self) -> None: legacy_payload = { "forecast": [{"time": 1, "p10": 1.0, "p50": 1.1, "p90": 1.2}], "forecast_candles": [], "display": {"mode": "raw_legacy_ohlc_p50"}, "ensemble": {"mode": "legacy_plus_anchor", "confidence": 55.0}, "model": { "input_semantics": { "feature_channels": ["open", "high", "low", "close", "volume", "amount"], "price_mode": "ohlc4_replicated_across_ohlc", "base_signal": "ohlc4", "volume_mode": "forced_zero", "amount_mode": "forced_zero", "active_forecast_channels": ["ohlc4"], } }, } self.assertFalse(main._forecast_payload_is_current(legacy_payload)) def test_forecast_payload_schema_guard_accepts_current_multi_model_payload(self) -> None: current_payload = self._build_current_forecast_payload() self.assertTrue(main._forecast_payload_is_current(current_payload)) def test_forecast_payload_schema_guard_rejects_legacy_forecast_candles_field(self) -> None: stale_payload = self._build_current_forecast_payload() stale_payload["forecast_candles"] = [] self.assertFalse(main._forecast_payload_is_current(stale_payload)) def test_finalize_forecast_error_payload_omits_legacy_forecast_candles_field(self) -> None: payload = { "symbol": self.symbol_a, "interval": "1h", "forecast_rows": [], "error": "offline", "path_checked": "test-path", "ai_runtime": {"mode": "local_only", "model": "offline"}, } response = asyncio.run(main._finalize_forecast_response_payload(payload)) self.assertNotIn("forecast_candles", response) self.assertEqual(response["display"]["output_mode"], "single_future_ohlc4_line") self.assertEqual(response["display"]["forecast_reference_field"], "last_ohlc4") self.assertEqual(response["display"]["visual_anchor_field"], "last_close") def test_forecast_payload_schema_guard_rejects_missing_reference_fields(self) -> None: stale_payload = self._build_current_forecast_payload() stale_payload["display"] = { "mode": "multi_model_ohlc4_line", "channels": ["ohlc4"], "output_mode": "single_future_ohlc4_line", } self.assertFalse(main._forecast_payload_is_current(stale_payload)) def test_forecast_payload_schema_guard_rejects_missing_model_selection(self) -> None: stale_payload = self._build_current_forecast_payload() stale_payload.pop("model_selection", None) self.assertFalse(main._forecast_payload_is_current(stale_payload)) def test_forecast_payload_schema_guard_rejects_missing_forecast_models(self) -> None: stale_payload = self._build_current_forecast_payload() stale_payload.pop("forecast_models", None) self.assertFalse(main._forecast_payload_is_current(stale_payload)) def test_forecast_payload_schema_guard_rejects_missing_combination_mode(self) -> None: stale_payload = self._build_current_forecast_payload() stale_payload["display"].pop("combination_mode", None) self.assertFalse(main._forecast_payload_is_current(stale_payload)) def test_forecast_payload_schema_guard_rejects_enabled_model_without_success(self) -> None: stale_payload = self._build_current_forecast_payload() stale_payload["forecast_models"]["kronos"]["success"] = False self.assertFalse(main._forecast_payload_is_current(stale_payload)) def test_calc_ai_forecast_score_keeps_market_return_and_tracks_model_reference_return(self) -> None: score = main._calc_ai_forecast_score( blended={ "p10": [100.0, 101.0], "p50": [101.0, 102.0], "p90": [102.0, 103.0], "confidence": 60.0, "scale": 1.0, }, forecast_rows=[], last_close=100.0, indicators={"trend": {}, "atr": {"pct": 1.0}, "rsi": {"value": 50.0}}, horizon=2, interval="1h", model_reference_price=101.0, ) self.assertEqual(score["forecast_return_pct"], 2.0) self.assertAlmostEqual( score["model_reference_return_pct"], round(((102.0 - 101.0) / 101.0) * 100.0, 2), ) def test_synthetic_component_history_reuses_source_cache(self) -> None: calls: list[tuple[str, str, str, int]] = [] async def fake_fetch_from_source( source: str, symbol: str, interval: str, limit: int, ) -> list[dict[str, float]]: calls.append((source, symbol, interval, limit)) return [ {"time": 1, "open": 1.0, "high": 1.1, "low": 0.9, "close": 1.0, "volume": 0.0}, {"time": 2, "open": 1.0, "high": 1.2, "low": 0.95, "close": 1.1, "volume": 0.0}, {"time": 3, "open": 1.1, "high": 1.25, "low": 1.0, "close": 1.15, "volume": 0.0}, ] main.source_history_cache.clear() main._SOURCE_HISTORY_INFLIGHT.clear() try: with patch.object(main, "_fetch_historical_from_source", side_effect=fake_fetch_from_source): asyncio.run(main._build_synthetic_symbol_history("EURX", "1d", 3, "synthetic-eurx-1")) asyncio.run(main._build_synthetic_symbol_history("EURX", "1d", 3, "synthetic-eurx-2")) finally: main.source_history_cache.clear() main._SOURCE_HISTORY_INFLIGHT.clear() unique_component_symbols = {item[1] for item in calls} self.assertEqual(len(calls), len(unique_component_symbols)) self.assertEqual(unique_component_symbols, {"AUDUSD", "EURGBP", "EURJPY", "EURUSD", "NZDUSD", "USDCAD", "USDCHF"}) def test_ttl_cache_returns_defensive_copy(self) -> None: cache = main.TTLCache() payload = {"forecast": [{"price": 100.0}], "meta": {"source": "memory"}} cache.set("sample", payload, ttl_seconds=30) payload["forecast"][0]["price"] = 999.0 payload["meta"]["source"] = "mutated" cached_once = cache.get("sample") self.assertEqual(cached_once["forecast"][0]["price"], 100.0) self.assertEqual(cached_once["meta"]["source"], "memory") cached_once["forecast"][0]["price"] = 555.0 cached_twice = cache.get("sample") self.assertEqual(cached_twice["forecast"][0]["price"], 100.0) def test_persistent_cache_queue_copies_payload_before_enqueue(self) -> None: class FakeQueue: def __init__(self) -> None: self.item = None def put_nowait(self, item: tuple[str, object, int]) -> None: self.item = item fd, temp_path = tempfile.mkstemp(suffix=".db") os.close(fd) try: cache = main.PersistentCache( db_path=temp_path, cache_version_getter=lambda: "test-version", logger=main.logger, ) queue = FakeQueue() cache._queue = queue payload = {"nested": {"value": 1}} cache.set("queued", payload, ttl=60) payload["nested"]["value"] = 7 self.assertIsNotNone(queue.item) _, queued_payload, queued_ttl = queue.item self.assertEqual(queued_ttl, 60) self.assertEqual(queued_payload["nested"]["value"], 1) finally: if os.path.exists(temp_path): try: os.remove(temp_path) except PermissionError: pass if __name__ == "__main__": unittest.main()