yuqei
450 lines
20 KiB
Python
450 lines
20 KiB
Python
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from src.api import get_health
|
|
from src.api.backtest import get_recent_backtest_summaries, normalize_backtest_summary_row
|
|
from src.api.server import route_request
|
|
from src.api.signals import get_recent_signals
|
|
from src.services.backtest import BacktestCostConfig, BacktestQualityGate
|
|
from src.services.execution import AccountRiskState
|
|
from src.main import run_closed_loop
|
|
|
|
|
|
class FakeResult:
|
|
def __init__(self, rows):
|
|
self._rows = rows
|
|
|
|
def mappings(self):
|
|
return self
|
|
|
|
def all(self):
|
|
return self._rows
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, rows):
|
|
self.rows = rows
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return None
|
|
|
|
def execute(self, sql, params):
|
|
return FakeResult(self.rows)
|
|
|
|
|
|
class ApiTests(unittest.TestCase):
|
|
def test_health_returns_expected_fields(self) -> None:
|
|
health = get_health()
|
|
|
|
self.assertEqual(health["app"], "ai-ict")
|
|
self.assertIn("database_configured", health)
|
|
self.assertIn("session_timezone", health)
|
|
|
|
def test_get_recent_signals_uses_session_query(self) -> None:
|
|
rows = [{"id": 1, "side": "buy", "status": "awaiting_confirmation"}]
|
|
with patch("src.api.signals.SessionLocal", return_value=FakeSession(rows)):
|
|
result = get_recent_signals(instrument_id=1, limit=1)
|
|
|
|
self.assertEqual(result, rows)
|
|
|
|
def test_get_recent_backtest_summaries_uses_session_query(self) -> None:
|
|
rows = [{"id": 10, "summary": {"total_signals": 3}}]
|
|
with patch("src.api.backtest.SessionLocal", return_value=FakeSession(rows)):
|
|
result = get_recent_backtest_summaries(instrument_id=1, limit=1)
|
|
|
|
self.assertEqual(result[0]["id"], 10)
|
|
self.assertIn("quality_gate", result[0]["summary"])
|
|
self.assertIn("quality_diagnostics", result[0]["summary"])
|
|
|
|
def test_normalize_backtest_summary_row_preserves_existing_quality_gate(self) -> None:
|
|
row = {"id": 10, "summary": {"total_signals": 3, "quality_gate": {"passed": True, "reasons": []}}}
|
|
|
|
result = normalize_backtest_summary_row(row)
|
|
|
|
self.assertEqual(result["summary"]["quality_gate"], {"passed": True, "reasons": []})
|
|
self.assertEqual(result["summary"]["quality_diagnostics"]["quality_state"], "passed")
|
|
|
|
def test_route_request_signals_returns_items_payload(self) -> None:
|
|
status, payload = route_request(
|
|
"/signals",
|
|
{"instrument_id": ["1"], "limit": ["1"]},
|
|
dependencies={"signals": lambda instrument_id, limit: [{"id": 1}], "health": lambda: {}, "candles": lambda **kwargs: [], "sessions": lambda **kwargs: [], "backtests": lambda **kwargs: []},
|
|
)
|
|
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(payload, {"items": [{"id": 1}]})
|
|
|
|
def test_route_request_signals_requires_instrument_id(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "missing_required_query_param:instrument_id"):
|
|
route_request(
|
|
"/signals",
|
|
{},
|
|
dependencies={"signals": lambda instrument_id, limit: [], "health": lambda: {}, "candles": lambda **kwargs: [], "sessions": lambda **kwargs: [], "backtests": lambda **kwargs: []},
|
|
)
|
|
|
|
|
|
class FakeSessionGenerator:
|
|
timezone = __import__("datetime").timezone.utc
|
|
|
|
def generate_for_date(self, instrument_id, timeframe, session_date):
|
|
return 5
|
|
|
|
|
|
class FakeSwingService:
|
|
def detect_swings(self, **kwargs):
|
|
return [type("Swing", (), {"kind": "swing_high", "price": 101.0, "ts": "t", "strength": "ST", "left_bars": 2, "right_bars": 2, "confirmed": True})()]
|
|
|
|
def persist_swings(self, **kwargs):
|
|
return 1
|
|
|
|
|
|
class FakeRangeService:
|
|
def build_active_range(self, **kwargs):
|
|
return type("Range", (), {"high": 110.0, "low": 90.0, "equilibrium": 100.0, "high_swing_id": 1, "low_swing_id": 2, "ts_start": "t"})()
|
|
|
|
def persist_active_range(self, **kwargs):
|
|
return 1
|
|
|
|
|
|
class FakeLiquidityService:
|
|
def build_candidates(self, **kwargs):
|
|
return [type("Pool", (), {"side": "buy_side", "pool_type": "prior_high", "price_low": 101.0, "price_high": 101.0, "reference_ids": [1], "created_ts": "t", "meta": {}})()]
|
|
|
|
def persist_candidates(self, **kwargs):
|
|
return 1
|
|
|
|
|
|
class FakeSweepService:
|
|
def detect_sweeps(self, **kwargs):
|
|
return [type("Sweep", (), {"side_swept": "buy_side", "pool_id": 1, "sweep_high": 102.0, "sweep_low": 99.0, "sweep_ts": "t", "close_back_inside": True, "validation_level": "validated", "status": "confirmed", "meta": {}})()]
|
|
|
|
def persist_sweeps(self, **kwargs):
|
|
return 1
|
|
|
|
|
|
class FakeDisplacementService:
|
|
def detect_displacement(self, **kwargs):
|
|
return [type("Disp", (), {"direction": "bullish", "start_ts": "t1", "end_ts": "t2", "strength_score": 0.8, "body_ratio": 2.0, "has_fvg": True, "is_valid": True})()]
|
|
|
|
def persist_displacement(self, **kwargs):
|
|
return 1
|
|
|
|
|
|
class FakeMSSService:
|
|
def detect_mss(self, **kwargs):
|
|
return [type("MSS", (), {"direction": "bullish", "broken_level": 100.0, "reference_swing_id": 1, "displacement_event_id": 2, "accepted": True, "ts": "t", "meta": {}})()]
|
|
|
|
def persist_mss(self, **kwargs):
|
|
return 1
|
|
|
|
|
|
class FakeFVGService:
|
|
def detect_fvg(self, **kwargs):
|
|
return [type("FVG", (), {"direction": "bullish", "upper": 103.0, "lower": 101.0, "start_ts": "t1", "end_ts": "t2", "fill_pct": 0.0, "status": "open", "related_displacement_id": 2, "meta": {}})()]
|
|
|
|
def persist_fvg(self, **kwargs):
|
|
return 1
|
|
|
|
def update_existing_fvgs(self, **kwargs):
|
|
return 1
|
|
|
|
|
|
class FakeBiasService:
|
|
def __init__(self):
|
|
self.persist_calls = 0
|
|
|
|
def build_bias(self, **kwargs):
|
|
return type("Bias", (), {"daily_bias": "bullish", "confidence": 0.7, "execution_intensity": "high", "narrative_code": "bullish_structure_shift", "draw_side": "buy_side", "range_id": 1, "reason_codes": ["latest_mss"], "snapshot_ts": "t"})()
|
|
|
|
def persist_bias(self, **kwargs):
|
|
self.persist_calls += 1
|
|
return 11
|
|
|
|
|
|
class FakeSignalService:
|
|
def __init__(self):
|
|
self.persist_calls = 0
|
|
|
|
def build_signal(self, **kwargs):
|
|
return type("Signal", (), {"side": "buy", "status": "awaiting_confirmation", "entry_low": 101.0, "entry_high": 103.0, "stop_loss": 99.0, "tp1": 104.0, "tp2": 106.0, "rr_tp1": 1.0, "rr_tp2": 2.0, "session_code": "LONDON", "created_ts": "t", "expires_ts": "t2", "invalidations": [], "evidence": {}, "meta": {"execution_permission": True}})()
|
|
|
|
def persist_signal(self, **kwargs):
|
|
self.persist_calls += 1
|
|
return 1
|
|
|
|
|
|
class FakeNoNewSignalService(FakeSignalService):
|
|
def persist_signal(self, **kwargs):
|
|
self.persist_calls += 1
|
|
return 0
|
|
|
|
|
|
class FakeExecutionService:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
self.settlement_calls = 0
|
|
self.last_create_kwargs = None
|
|
|
|
def create_execution(self, **kwargs):
|
|
self.calls += 1
|
|
self.last_create_kwargs = kwargs
|
|
return 99
|
|
|
|
def settle_latest_execution_for_signal(self, **kwargs):
|
|
self.settlement_calls += 1
|
|
return {"id": 99, "status": "closed", "meta": {"lifecycle_state": "tp2_closed", "risk_checks": []}}
|
|
|
|
def persist_execution_settlement(self, **kwargs):
|
|
self.settlement_calls += 1
|
|
return {"id": 99, "status": "closed", "meta": {"lifecycle_state": "tp2_closed", "risk_checks": []}}
|
|
|
|
|
|
class FakeBacktestService:
|
|
def __init__(self):
|
|
self.last_kwargs = None
|
|
|
|
def run_backtest(self, **kwargs):
|
|
self.last_kwargs = kwargs
|
|
outcome = type("Outcome", (), {"signal_id": 1, "outcome": "win", "mfe": 3.0, "mae": 1.0, "tp1_hit": True, "tp2_hit": True, "invalidated_before_entry": False, "error_tags": [], "meta": {"signal_side": "buy", "session_codes": ["LONDON"], "weekday": "Monday"}})()
|
|
return 55, [outcome]
|
|
|
|
|
|
class FakePreMarketReportService:
|
|
def build_report(self, **kwargs):
|
|
return {"report_type": "pre_market_plan"}
|
|
|
|
|
|
class FakeIntradayReportService:
|
|
def build_report(self, **kwargs):
|
|
return {"report_type": "intraday_explanation"}
|
|
|
|
|
|
class FakePostTradeReviewService:
|
|
def build_report(self, **kwargs):
|
|
return {"report_type": "post_trade_review"}
|
|
|
|
|
|
class ClosedLoopTests(unittest.TestCase):
|
|
def test_run_closed_loop_returns_summary_and_reports(self) -> None:
|
|
result = run_closed_loop(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
backtest_limit=10,
|
|
run_execution=True,
|
|
dependencies={
|
|
"session_generator": FakeSessionGenerator(),
|
|
"swing_service": FakeSwingService(),
|
|
"range_service": FakeRangeService(),
|
|
"liquidity_service": FakeLiquidityService(),
|
|
"sweep_service": FakeSweepService(),
|
|
"displacement_service": FakeDisplacementService(),
|
|
"mss_service": FakeMSSService(),
|
|
"fvg_service": FakeFVGService(),
|
|
"bias_service": FakeBiasService(),
|
|
"signal_service": FakeSignalService(),
|
|
"execution_service": FakeExecutionService(),
|
|
"backtest_service": FakeBacktestService(),
|
|
"pre_market_report_service": FakePreMarketReportService(),
|
|
"intraday_report_service": FakeIntradayReportService(),
|
|
"post_trade_review_service": FakePostTradeReviewService(),
|
|
"get_recent_signals": lambda instrument_id, limit: [{"id": 1, "session_code": "LONDON", "status": "awaiting_confirmation", "side": "buy", "invalidations": []}],
|
|
"get_recent_backtest_summaries": lambda instrument_id, limit: [{"id": 55, "summary": {"total_signals": 1}}],
|
|
},
|
|
)
|
|
|
|
self.assertEqual(result["writes"]["sessions"], 5)
|
|
self.assertEqual(result["writes"]["execution_id"], 99)
|
|
self.assertEqual(result["writes"]["backtest_run_id"], 55)
|
|
self.assertEqual(result["reports"]["pre_market"]["report_type"], "pre_market_plan")
|
|
self.assertEqual(result["reports"]["intraday"]["report_type"], "intraday_explanation")
|
|
self.assertEqual(result["reports"]["post_trade"]["report_type"], "post_trade_review")
|
|
self.assertEqual(result["writes"]["execution_status"], "closed")
|
|
self.assertEqual(result["writes"]["execution_lifecycle_state"], "tp2_closed")
|
|
self.assertIn("backtest_quality_gate", result)
|
|
self.assertFalse(result["backtest_quality_gate"]["passed"])
|
|
self.assertIn("backtest_quality_diagnostics", result)
|
|
self.assertEqual(result["backtest_quality_diagnostics"]["quality_state"], "blocked")
|
|
|
|
def test_run_closed_loop_does_not_execute_stale_latest_signal_without_new_signal(self) -> None:
|
|
execution_service = FakeExecutionService()
|
|
|
|
result = run_closed_loop(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
backtest_limit=10,
|
|
run_execution=True,
|
|
dependencies={
|
|
"session_generator": FakeSessionGenerator(),
|
|
"swing_service": FakeSwingService(),
|
|
"range_service": FakeRangeService(),
|
|
"liquidity_service": FakeLiquidityService(),
|
|
"sweep_service": FakeSweepService(),
|
|
"displacement_service": FakeDisplacementService(),
|
|
"mss_service": FakeMSSService(),
|
|
"fvg_service": FakeFVGService(),
|
|
"bias_service": FakeBiasService(),
|
|
"signal_service": FakeNoNewSignalService(),
|
|
"execution_service": execution_service,
|
|
"backtest_service": FakeBacktestService(),
|
|
"pre_market_report_service": FakePreMarketReportService(),
|
|
"intraday_report_service": FakeIntradayReportService(),
|
|
"post_trade_review_service": FakePostTradeReviewService(),
|
|
"get_recent_signals": lambda instrument_id, limit: [{"id": 7, "session_code": "LONDON", "status": "awaiting_confirmation", "side": "buy", "invalidations": []}],
|
|
"get_recent_backtest_summaries": lambda instrument_id, limit: [{"id": 55, "summary": {"total_signals": 1}}],
|
|
},
|
|
)
|
|
|
|
self.assertEqual(execution_service.calls, 0)
|
|
self.assertEqual(execution_service.settlement_calls, 0)
|
|
self.assertIsNone(result["writes"]["execution_id"])
|
|
self.assertEqual(result["writes"]["execution_status"], "skipped")
|
|
self.assertEqual(result["writes"]["execution_lifecycle_state"], "skipped")
|
|
|
|
def test_run_closed_loop_repeat_uses_service_idempotency_contracts(self) -> None:
|
|
bias_service = FakeBiasService()
|
|
signal_service = FakeSignalService()
|
|
execution_service = FakeExecutionService()
|
|
recent_signals = [{"id": 1, "session_code": "LONDON", "status": "awaiting_confirmation", "side": "buy", "invalidations": []}]
|
|
|
|
deps = {
|
|
"session_generator": FakeSessionGenerator(),
|
|
"swing_service": FakeSwingService(),
|
|
"range_service": FakeRangeService(),
|
|
"liquidity_service": FakeLiquidityService(),
|
|
"sweep_service": FakeSweepService(),
|
|
"displacement_service": FakeDisplacementService(),
|
|
"mss_service": FakeMSSService(),
|
|
"fvg_service": FakeFVGService(),
|
|
"bias_service": bias_service,
|
|
"signal_service": signal_service,
|
|
"execution_service": execution_service,
|
|
"backtest_service": FakeBacktestService(),
|
|
"pre_market_report_service": FakePreMarketReportService(),
|
|
"intraday_report_service": FakeIntradayReportService(),
|
|
"post_trade_review_service": FakePostTradeReviewService(),
|
|
"get_recent_signals": lambda instrument_id, limit: recent_signals,
|
|
"get_recent_backtest_summaries": lambda instrument_id, limit: [{"id": 55, "summary": {"total_signals": 1}}],
|
|
}
|
|
|
|
first = run_closed_loop(instrument_id=1, timeframe="1m", backtest_limit=10, run_execution=True, dependencies=deps)
|
|
second = run_closed_loop(instrument_id=1, timeframe="1m", backtest_limit=10, run_execution=True, dependencies=deps)
|
|
|
|
self.assertEqual(first["writes"]["bias_snapshot_id"], 11)
|
|
self.assertEqual(second["writes"]["bias_snapshot_id"], 11)
|
|
self.assertEqual(bias_service.persist_calls, 2)
|
|
self.assertEqual(signal_service.persist_calls, 2)
|
|
self.assertEqual(execution_service.calls, 2)
|
|
self.assertEqual(execution_service.settlement_calls, 2)
|
|
|
|
def test_run_closed_loop_passes_backtest_cost_config(self) -> None:
|
|
backtest_service = FakeBacktestService()
|
|
cost_config = BacktestCostConfig(fee_rate=0.0005, slippage_rate=0.0005, fixed_cost_r=0.05)
|
|
deps = {
|
|
"session_generator": FakeSessionGenerator(),
|
|
"swing_service": FakeSwingService(),
|
|
"range_service": FakeRangeService(),
|
|
"liquidity_service": FakeLiquidityService(),
|
|
"sweep_service": FakeSweepService(),
|
|
"displacement_service": FakeDisplacementService(),
|
|
"mss_service": FakeMSSService(),
|
|
"fvg_service": FakeFVGService(),
|
|
"bias_service": FakeBiasService(),
|
|
"signal_service": FakeSignalService(),
|
|
"execution_service": FakeExecutionService(),
|
|
"backtest_service": backtest_service,
|
|
"pre_market_report_service": FakePreMarketReportService(),
|
|
"intraday_report_service": FakeIntradayReportService(),
|
|
"post_trade_review_service": FakePostTradeReviewService(),
|
|
"get_recent_signals": lambda instrument_id, limit: [{"id": 1, "session_code": "LONDON", "status": "awaiting_confirmation", "side": "buy", "invalidations": []}],
|
|
"get_recent_backtest_summaries": lambda instrument_id, limit: [{"id": 55, "summary": {"total_signals": 1}}],
|
|
}
|
|
|
|
run_closed_loop(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
backtest_limit=10,
|
|
run_execution=True,
|
|
backtest_cost_config=cost_config,
|
|
dependencies=deps,
|
|
)
|
|
|
|
self.assertEqual(backtest_service.last_kwargs["cost_config"], cost_config)
|
|
|
|
def test_run_closed_loop_passes_backtest_quality_gate(self) -> None:
|
|
backtest_service = FakeBacktestService()
|
|
quality_gate = BacktestQualityGate(min_total_signals=1, min_profit_factor_r=1.0)
|
|
deps = {
|
|
"session_generator": FakeSessionGenerator(),
|
|
"swing_service": FakeSwingService(),
|
|
"range_service": FakeRangeService(),
|
|
"liquidity_service": FakeLiquidityService(),
|
|
"sweep_service": FakeSweepService(),
|
|
"displacement_service": FakeDisplacementService(),
|
|
"mss_service": FakeMSSService(),
|
|
"fvg_service": FakeFVGService(),
|
|
"bias_service": FakeBiasService(),
|
|
"signal_service": FakeSignalService(),
|
|
"execution_service": FakeExecutionService(),
|
|
"backtest_service": backtest_service,
|
|
"pre_market_report_service": FakePreMarketReportService(),
|
|
"intraday_report_service": FakeIntradayReportService(),
|
|
"post_trade_review_service": FakePostTradeReviewService(),
|
|
"get_recent_signals": lambda instrument_id, limit: [{"id": 1, "session_code": "LONDON", "status": "awaiting_confirmation", "side": "buy", "invalidations": []}],
|
|
"get_recent_backtest_summaries": lambda instrument_id, limit: [{"id": 55, "summary": {"total_signals": 1}}],
|
|
}
|
|
|
|
run_closed_loop(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
backtest_limit=10,
|
|
run_execution=True,
|
|
backtest_quality_gate=quality_gate,
|
|
dependencies=deps,
|
|
)
|
|
|
|
self.assertEqual(backtest_service.last_kwargs["quality_gate"], quality_gate)
|
|
|
|
def test_run_closed_loop_passes_execution_account_state(self) -> None:
|
|
execution_service = FakeExecutionService()
|
|
account_state = AccountRiskState(daily_pnl_pct=-0.01, consecutive_losses=1)
|
|
deps = {
|
|
"session_generator": FakeSessionGenerator(),
|
|
"swing_service": FakeSwingService(),
|
|
"range_service": FakeRangeService(),
|
|
"liquidity_service": FakeLiquidityService(),
|
|
"sweep_service": FakeSweepService(),
|
|
"displacement_service": FakeDisplacementService(),
|
|
"mss_service": FakeMSSService(),
|
|
"fvg_service": FakeFVGService(),
|
|
"bias_service": FakeBiasService(),
|
|
"signal_service": FakeSignalService(),
|
|
"execution_service": execution_service,
|
|
"backtest_service": FakeBacktestService(),
|
|
"pre_market_report_service": FakePreMarketReportService(),
|
|
"intraday_report_service": FakeIntradayReportService(),
|
|
"post_trade_review_service": FakePostTradeReviewService(),
|
|
"get_recent_signals": lambda instrument_id, limit: [{"id": 1, "session_code": "LONDON", "status": "awaiting_confirmation", "side": "buy", "invalidations": []}],
|
|
"get_recent_backtest_summaries": lambda instrument_id, limit: [{"id": 55, "summary": {"total_signals": 1}}],
|
|
}
|
|
|
|
run_closed_loop(
|
|
instrument_id=1,
|
|
timeframe="1m",
|
|
backtest_limit=10,
|
|
run_execution=True,
|
|
execution_account_state=account_state,
|
|
dependencies=deps,
|
|
)
|
|
|
|
self.assertEqual(execution_service.last_create_kwargs["account_state"], account_state)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|