yuqei/ai-exchange
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Review Now mainline gates

Browse Source
This commit is contained in:
Codex
2026-05-21 12:01:50 +08:00
parent af319e9895
commit 3c8613632f
12 changed files with 521 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_api_and_main.py
+17 -1
View File
@@ -6,7 +6,7 @@ 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.backtest import get_recent_backtest_summaries, get_strategy_backtest_surface, 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, BacktestReplayConfig
@@ -102,6 +102,22 @@ class ApiTests(unittest.TestCase):
self.assertEqual(result["ict_rulebook_mapping"]["surface"], "strategy_backtest")
self.assertIn("backtest_evidence", result["ict_rulebook_mapping"]["present_concepts"])
def test_strategy_backtest_surface_exposes_cockpit_promotion_gate(self) -> None:
rows = [
{
"id": 10,
"summary": {
"total_signals": 3,
"quality_gate": {"passed": False, "reasons": ["sample_size_too_small"]},
},
}
]
with patch("src.api.backtest.SessionLocal", return_value=FakeSession(rows)):
result = get_strategy_backtest_surface(instrument_id=1, limit=1)
self.assertEqual(result["cockpit_promotion_gate"]["status"], "blocked")
self.assertIn("sample_size_too_small", result["cockpit_promotion_gate"]["reasons"])
def test_route_request_trading_candidates_returns_items_payload(self) -> None:
status, payload = route_request(
"/trading/candidates",
tests/unit/test_documentation_consistency.py
+1
View File
@@ -22,6 +22,7 @@ ACTIVE_TOP_LEVEL_DOCS = {
"AI_ICT_实盘辅助分析决策助手ICT主线开发方案_2026-05-21_v1.md",
"AI_ICT_主线裁剪审计_S1_2026-05-21_v1.md",
"AI_ICT_ICT主线概念字典_S2_2026-05-21_v1.md",
"AI_ICT_ReviewNow主线Gate_S3S4_2026-05-21_v1.md",
"AI_ICT_主线纠偏与功能裁剪方案_2026-05-20_v1.md",
"AI_ICT_实盘辅助分析决策助手上线必做清单_2026-05-18_v1.md",
"AI_ICT_实盘辅助分析决策助手使用说明_2026-05-18_v1.md",
tests/unit/test_ict_mainline_taxonomy.py
+26
View File
@@ -4,6 +4,7 @@ from src.services.ict_mainline import (
ICT_MAINLINE_CHAIN,
build_ict_rulebook_mapping,
build_ict_rulebook_summary,
build_review_now_mainline_gate,
classify_ict_concepts,
get_ict_concept,
)
@@ -114,6 +115,31 @@ class IctMainlineTaxonomyTests(unittest.TestCase):
self.assertIn("session", concepts)
self.assertIn("no_trade_conditions", concepts)
def test_review_now_mainline_gate_requires_plan_and_backtest(self) -> None:
blocked = build_review_now_mainline_gate(
active_trading_plan=None,
backtest_quality_gate={"passed": False, "reasons": ["sample_size_too_small"]},
)
self.assertEqual(blocked["status"], "blocked")
self.assertFalse(blocked["review_now_allowed"])
self.assertIn("missing_active_trading_plan", blocked["blockers"])
self.assertIn("sample_size_too_small", blocked["blockers"])
passed = build_review_now_mainline_gate(
active_trading_plan={
"id": 10,
"status": "active",
"primary_draw": {"side": "buy_side"},
"invalidations": [{"code": "range_low_lost"}],
"no_trade_conditions": [],
},
backtest_quality_gate={"passed": True, "reasons": []},
)
self.assertEqual(passed["status"], "passed")
self.assertTrue(passed["review_now_allowed"])
if __name__ == "__main__":
unittest.main()
tests/unit/test_operator_workspace_service.py
+122
View File
@@ -129,6 +129,7 @@ class OperatorWorkspaceServiceTests(unittest.TestCase):
announcements_provider=_AnnouncementsProvider(),
external_chart_provider=_ExternalChartProvider(),
decision_assistant_repository=_DecisionAssistantRepo(),
market_state_repository=_BacktestQualityGateRepo(passed=True),
)
summary = service.build_summary(
@@ -195,6 +196,9 @@ class OperatorWorkspaceServiceTests(unittest.TestCase):
self.assertEqual(summary["top_opportunities"][0]["kind"], "focus_candidate")
self.assertTrue(any(item["kind"] == "decision_packet" for item in summary["top_priority_items"]))
self.assertEqual(summary["candidate_watchlist_tiers"]["review_now"][0]["candidate_id"], 18)
self.assertTrue(summary["candidate_watchlist_tiers"]["review_now"][0]["review_now_allowed"])
self.assertEqual(summary["mainline_focus_gate"]["status"], "passed")
self.assertTrue(summary["mainline_focus_gate"]["review_now_allowed"])
self.assertEqual(summary["candidate_watchlist_tiers"]["strong_watch"][0]["candidate_id"], 21)
self.assertEqual(summary["candidate_watchlist_tiers"]["soft_watch"][0]["candidate_id"], 22)
self.assertEqual(summary["candidate_watchlist_tiers"]["review_now"][0]["action_method"], "POST")
@@ -247,6 +251,8 @@ class OperatorWorkspaceServiceTests(unittest.TestCase):
self.assertIn("Read-only Live Analysis", trader_mode_summary["drilldown_sections"])
self.assertEqual(summary["evidence_summary"]["trader_mode_status"], "review_now")
self.assertEqual(summary["evidence_summary"]["trader_mode_color"], "review")
self.assertEqual(summary["evidence_summary"]["mainline_focus_gate_status"], "passed")
self.assertTrue(summary["evidence_summary"]["review_now_allowed_by_mainline"])
self.assertIn("focus_matrix", summary["candidate_watchlist_summary"])
self.assertIn("next_watchlist_focus", summary["candidate_watchlist_summary"])
self.assertIn("next_watchlist_focus_prompt", summary["candidate_watchlist_summary"])
@@ -817,6 +823,85 @@ class OperatorWorkspaceServiceTests(unittest.TestCase):
self.assertIn("action:observe_only->confirm_ticket", summary["changed_since_last"]["items"])
self.assertIn("candidate:none->18", summary["changed_since_last"]["items"])
def test_decision_assistant_blocks_review_now_without_active_plan(self) -> None:
service = DecisionAssistantSummaryService(
launch_readiness_service=_LaunchReadinessService(),
session_focus_service=_SessionFocusSummaryService(),
operator_digest_service=_OperatorDigestSummaryService(),
decision_repository=_PreviousPacketRepo(),
candidate_repository=_NoActivePlanCandidateWatchlistRepo(),
research_council_service=_ResearchCouncilSummaryService(),
exchange_status_provider=_ExchangeStatusProvider(),
announcements_provider=_AnnouncementsProvider(),
external_chart_provider=_ExternalChartProvider(),
decision_assistant_repository=_DecisionAssistantRepo(),
market_state_repository=_BacktestQualityGateRepo(passed=True),
)
summary = service.build_summary(
instrument_id=1,
timeframe="5m",
decision_packet={
"status": "actionable",
"action_recommendation": "confirm_ticket",
"candidate_id": 18,
"packet_ts": datetime.now(timezone.utc).isoformat(),
"setup_type": "SWEEP_MSS_FVG",
"why_allowed": ["candidate_confirmed"],
"why_blocked": [],
},
weekly_review={"next_week_actions": [], "session_performance": {}},
recent_alerts=[],
recent_notes=[],
recent_executions=[],
)
self.assertEqual(summary["mainline_focus_gate"]["status"], "blocked")
self.assertIn("missing_active_trading_plan", summary["mainline_focus_gate"]["blockers"])
self.assertEqual(summary["candidate_watchlist_tiers"]["review_now"], [])
self.assertEqual(summary["candidate_watchlist_tiers"]["strong_watch"][0]["candidate_id"], 18)
self.assertIn("missing_active_trading_plan", summary["candidate_watchlist_tiers"]["strong_watch"][0]["mainline_gate_blockers"])
self.assertEqual(summary["trader_mode_summary"]["status"], "observe_only")
self.assertEqual(summary["trader_mode_summary"]["status_label"], "Observe Only")
def test_decision_assistant_blocks_review_now_when_backtest_quality_gate_fails(self) -> None:
service = DecisionAssistantSummaryService(
launch_readiness_service=_LaunchReadinessService(),
session_focus_service=_SessionFocusSummaryService(),
operator_digest_service=_OperatorDigestSummaryService(),
decision_repository=_PreviousPacketRepo(),
candidate_repository=_CandidateWatchlistRepo(),
research_council_service=_ResearchCouncilSummaryService(),
exchange_status_provider=_ExchangeStatusProvider(),
announcements_provider=_AnnouncementsProvider(),
external_chart_provider=_ExternalChartProvider(),
decision_assistant_repository=_DecisionAssistantRepo(),
market_state_repository=_BacktestQualityGateRepo(passed=False, reasons=["sample_size_too_small"]),
)
summary = service.build_summary(
instrument_id=1,
timeframe="5m",
decision_packet={
"status": "actionable",
"action_recommendation": "confirm_ticket",
"candidate_id": 18,
"packet_ts": datetime.now(timezone.utc).isoformat(),
"setup_type": "SWEEP_MSS_FVG",
"why_allowed": ["candidate_confirmed"],
"why_blocked": [],
},
weekly_review={"next_week_actions": [], "session_performance": {}},
recent_alerts=[],
recent_notes=[],
recent_executions=[],
)
self.assertEqual(summary["mainline_focus_gate"]["status"], "blocked")
self.assertIn("sample_size_too_small", summary["mainline_focus_gate"]["blockers"])
self.assertEqual(summary["candidate_watchlist_tiers"]["review_now"], [])
self.assertEqual(summary["trader_mode_summary"]["status"], "observe_only")
self.assertIn("sample_size_too_small", summary["trader_mode_summary"]["key_reasons"])
def test_decision_assistant_trader_mode_freezes_stale_payload(self) -> None:
service = DecisionAssistantSummaryService(
@@ -830,6 +915,7 @@ class OperatorWorkspaceServiceTests(unittest.TestCase):
announcements_provider=_AnnouncementsProvider(),
external_chart_provider=_ExternalChartProvider(),
decision_assistant_repository=_DecisionAssistantRepo(),
market_state_repository=_BacktestQualityGateRepo(passed=True),
)
summary = service.build_summary(
@@ -873,6 +959,7 @@ class OperatorWorkspaceServiceTests(unittest.TestCase):
announcements_provider=_AnnouncementsProvider(),
external_chart_provider=_ExternalChartProvider(),
decision_assistant_repository=_DecisionAssistantRepo(),
market_state_repository=_BacktestQualityGateRepo(passed=True),
)
summary = service.build_summary(
@@ -1180,6 +1267,7 @@ class _CandidateWatchlistRepo:
return [
{
"id": 18,
"trading_plan_id": 101,
"session_code": "LONDON",
"status": "candidate",
"quality_score": 81.0,
@@ -1189,6 +1277,7 @@ class _CandidateWatchlistRepo:
},
{
"id": 21,
"trading_plan_id": 101,
"session_code": "LONDON",
"status": "candidate",
"quality_score": 72.0,
@@ -1198,6 +1287,7 @@ class _CandidateWatchlistRepo:
},
{
"id": 22,
"trading_plan_id": 101,
"session_code": "NY_AM",
"status": "candidate",
"quality_score": 44.0,
@@ -1207,6 +1297,7 @@ class _CandidateWatchlistRepo:
},
{
"id": 23,
"trading_plan_id": 101,
"session_code": "OFF_HOURS",
"status": "candidate",
"quality_score": 12.0,
@@ -1222,6 +1313,37 @@ class _CandidateWatchlistRepo:
return {"instrument_id": 1, **item}
return None
def fetch_active_trading_plan(self, *, instrument_id: int, plan_date=None):
return {
"id": 101,
"instrument_id": instrument_id,
"plan_date": "2026-05-12",
"status": "active",
"primary_draw": {"side": "buy_side", "source": "liquidity_pool"},
"invalidations": [{"code": "range_low_lost"}],
"no_trade_conditions": [],
"bias": "bullish",
}
class _NoActivePlanCandidateWatchlistRepo(_CandidateWatchlistRepo):
def fetch_active_trading_plan(self, *, instrument_id: int, plan_date=None):
return None
class _BacktestQualityGateRepo:
def __init__(self, *, passed: bool, reasons=None) -> None:
self.passed = passed
self.reasons = list(reasons or [])
def fetch_latest_backtest_quality_gate(self, *, instrument_id: int):
return {
"passed": self.passed,
"reasons": self.reasons,
"backtest_run_id": 489,
"observed_ts": "2026-05-12T01:00:00+00:00",
}
class _PreviousPacketRepo:
def fetch_latest_decision_packet(self, *, instrument_id: int):