Codex
91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
from pathlib import Path
|
|
import json
|
|
import os
|
|
import sys
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.services.reports import (
|
|
BacktestReportService,
|
|
IntradayReportService,
|
|
PostTradeReviewService,
|
|
PreMarketReportService,
|
|
WeeklyReviewService,
|
|
)
|
|
|
|
|
|
def main(env=None) -> int:
|
|
env = os.environ if env is None else env
|
|
output_json = _get_bool(env, "REPORTS_OUTPUT_JSON", False)
|
|
pre_market = 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",
|
|
)
|
|
intraday = IntradayReportService().build_report(
|
|
sweep_event={"side_swept": "buy_side", "close_back_inside": True},
|
|
structure_event={"event_type": "MSS", "direction": "bullish", "accepted": True},
|
|
fvg_zone={"direction": "bullish", "status": "open", "fill_pct": 0.0, "meta": {}},
|
|
trade_signal={"status": "awaiting_confirmation", "side": "buy", "session_code": "LONDON", "invalidations": []},
|
|
)
|
|
review = PostTradeReviewService().build_report(
|
|
trade_execution={"status": "open", "meta": {"risk_checks": []}},
|
|
trade_signal={"status": "awaiting_confirmation", "invalidations": []},
|
|
backtest_result={"outcome": "win", "invalidated_before_entry": False, "error_tags": []},
|
|
)
|
|
backtest_summary = BacktestReportService().build_summary([])
|
|
try:
|
|
weekly_review = WeeklyReviewService().build_summary()
|
|
except SQLAlchemyError:
|
|
weekly_review = {
|
|
"entry_count": 0,
|
|
"framework_execution_rate": 0.0,
|
|
"out_of_model_trade_rate": 0.0,
|
|
"risk_violation_count": 0,
|
|
"high_quality_session": None,
|
|
"low_quality_session": None,
|
|
"session_performance": {},
|
|
"next_week_actions": ["collect_more_executed_samples"],
|
|
}
|
|
|
|
payload = {
|
|
"passed": True,
|
|
"pre_market": pre_market,
|
|
"intraday": intraday,
|
|
"post_trade": review,
|
|
"backtest_summary": backtest_summary,
|
|
"weekly_review": weekly_review,
|
|
}
|
|
if output_json:
|
|
print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
|
|
return 0
|
|
|
|
print(f"Pre-market permission: {pre_market['execution_permission']}")
|
|
print(f"Intraday valid: {intraday['signal']['is_valid']}")
|
|
print(f"Post-trade aligned: {review['rule_aligned']}")
|
|
print(f"Backtest total signals: {backtest_summary['total_signals']}")
|
|
print(f"Backtest quality gate passed: {backtest_summary['quality_gate']['passed']}")
|
|
print(f"Weekly next actions: {weekly_review['next_week_actions']}")
|
|
return 0
|
|
|
|
|
|
def _get_bool(env, key: str, default: bool) -> bool:
|
|
value = env.get(key)
|
|
if value is None or value == "":
|
|
return default
|
|
return str(value).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|