yuqei
93 lines
4.1 KiB
Python
93 lines
4.1 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import unittest
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from src.services.reports import IntradayReportService, PostTradeReviewService, PreMarketReportService
|
|
from scripts import run_reports
|
|
|
|
|
|
class PreMarketReportTests(unittest.TestCase):
|
|
def test_builds_pre_market_plan(self) -> None:
|
|
report = PreMarketReportService().build_report(
|
|
bias_snapshot={
|
|
"daily_bias": "bullish",
|
|
"draw_side": "buy_side",
|
|
"execution_intensity": "high",
|
|
"reason_codes": ["latest_mss", "active_range_available"],
|
|
},
|
|
dealing_range={"id": 1, "high": 110.0, "low": 90.0, "equilibrium": 100.0},
|
|
current_price=98.0,
|
|
session_code="LONDON",
|
|
)
|
|
|
|
self.assertEqual(report["report_type"], "pre_market_plan")
|
|
self.assertEqual(report["daily_bias"], "bullish")
|
|
self.assertEqual(report["working_range"]["location"], "discount")
|
|
self.assertTrue(report["execution_permission"])
|
|
|
|
|
|
class IntradayReportTests(unittest.TestCase):
|
|
def test_builds_intraday_explanation_and_invalidations(self) -> None:
|
|
report = IntradayReportService().build_report(
|
|
sweep_event={"side_swept": "buy_side", "close_back_inside": True},
|
|
structure_event={"event_type": "MSS", "direction": "bearish", "accepted": True},
|
|
fvg_zone={"direction": "bearish", "status": "mitigated", "fill_pct": 1.0, "meta": {}},
|
|
trade_signal={"status": "awaiting_confirmation", "side": "sell", "session_code": "NY_AM", "invalidations": ["session_mismatch"]},
|
|
)
|
|
|
|
self.assertEqual(report["report_type"], "intraday_explanation")
|
|
self.assertFalse(report["signal"]["is_valid"])
|
|
self.assertIn("fvg_no_longer_actionable", report["invalidations"])
|
|
self.assertIn("session_mismatch", report["invalidations"])
|
|
|
|
|
|
class PostTradeReviewTests(unittest.TestCase):
|
|
def test_builds_post_trade_review(self) -> None:
|
|
report = PostTradeReviewService().build_report(
|
|
trade_execution={"status": "rejected", "meta": {"risk_checks": ["session_not_allowed"]}},
|
|
trade_signal={"status": "awaiting_context", "invalidations": ["session_mismatch"]},
|
|
backtest_result={"outcome": "loss", "invalidated_before_entry": False, "error_tags": ["same_candle_entry_exit_ambiguous"]},
|
|
)
|
|
|
|
self.assertEqual(report["report_type"], "post_trade_review")
|
|
self.assertFalse(report["rule_aligned"])
|
|
self.assertIn("execution_rejected", report["error_types"])
|
|
self.assertIn("session_not_allowed", report["error_types"])
|
|
self.assertEqual(report["execution_audit"]["audit_status"], "review")
|
|
self.assertEqual(report["execution_audit"]["lifecycle_state"], "rejected")
|
|
self.assertIn("inspect_stop_and_structure_context", report["improvement_notes"])
|
|
self.assertIn("review_same_candle_ambiguity_cases", report["improvement_notes"])
|
|
|
|
|
|
class ReportScriptTests(unittest.TestCase):
|
|
def test_run_reports_prints_backtest_quality_gate(self) -> None:
|
|
output = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_reports.main()
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertIn("Backtest quality gate passed: False", output.getvalue())
|
|
|
|
def test_run_reports_outputs_json_summary(self) -> None:
|
|
output = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_reports.main(env={"REPORTS_OUTPUT_JSON": "1"})
|
|
|
|
payload = json.loads(output.getvalue())
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertTrue(payload["passed"])
|
|
self.assertEqual(payload["pre_market"]["report_type"], "pre_market_plan")
|
|
self.assertEqual(payload["intraday"]["report_type"], "intraday_explanation")
|
|
self.assertEqual(payload["post_trade"]["report_type"], "post_trade_review")
|
|
self.assertIn("quality_gate", payload["backtest_summary"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|