from __future__ import annotations import argparse import json from collections import Counter, defaultdict from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from typing import Any STRATEGIES = ( "candidate_v76_fee_aware_anti_chase", "candidate_v76_strict_live_candidate", "candidate_v76_research_probe", "candidate_swing_trend_retest_research", ) def _read_jsonl_tail(path: Path, *, limit: int = 5000) -> list[dict[str, Any]]: if not path.exists(): return [] rows: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8", errors="ignore").splitlines()[-limit:]: try: row = json.loads(line) except json.JSONDecodeError: continue if isinstance(row, dict): rows.append(row) return rows def _d(value: Any, default: str = "0") -> Decimal: try: if value is None: return Decimal(default) return Decimal(str(value)) except Exception: return Decimal(default) def _classify_blocker(reason: str) -> str: if reason.startswith("confluence_"): return "confluence" if reason.startswith("tv_"): return "tradingview" if reason.startswith("market_"): return "market_regime" if "liquidity" in reason or "spread" in reason or "impact" in reason: return "liquidity" if "funding" in reason or "premium" in reason or "crowd" in reason or "open_interest" in reason: return "derivatives" if "risk" in reason or "quality" in reason or "reconcile" in reason: return "risk_quality" if "retest" in reason or "chase" in reason or "reclaim" in reason or "momentum" in reason: return "setup_quality" return "other" def summarize_strategy(strategy_id: str, *, runtime_root: Path, limit: int) -> dict[str, Any]: runtime_dir = runtime_root / strategy_id signals = _read_jsonl_tail(runtime_dir / "signal_journal.jsonl", limit=limit) trades = _read_jsonl_tail(runtime_dir / "trade_journal.jsonl", limit=limit) blocker_counts: Counter[str] = Counter() category_counts: Counter[str] = Counter() final_counts: Counter[str] = Counter() confluence_scores: list[Decimal] = [] tv_available = 0 would_enter = 0 for row in signals: final_counts[str(row.get("final_decision") or "unknown")] += 1 if row.get("would_enter"): would_enter += 1 reasons = row.get("block_reason") or [] if isinstance(reasons, str): reasons = [reasons] for reason in reasons: reason_s = str(reason) blocker_counts[reason_s] += 1 category_counts[_classify_blocker(reason_s)] += 1 conf = row.get("confluence_gate") if isinstance(row.get("confluence_gate"), dict) else {} if conf and conf.get("score") is not None: confluence_scores.append(_d(conf.get("score"))) tv_obj = row.get("tradingview_context") tv = tv_obj if isinstance(tv_obj, dict) else {} if tv.get("available"): tv_available += 1 closed = [row for row in trades if row.get("event") == "exit"] entries = [row for row in trades if row.get("event") == "entry"] pnl = sum((_d(row.get("net_pnl_usd", row.get("realized_pnl_usd"))) for row in closed), Decimal("0")) wins = sum(1 for row in closed if _d(row.get("net_pnl_usd", row.get("realized_pnl_usd"))) > 0) losses = sum(1 for row in closed if _d(row.get("net_pnl_usd", row.get("realized_pnl_usd"))) < 0) avg_conf = (sum(confluence_scores, Decimal("0")) / Decimal(len(confluence_scores))).quantize(Decimal("0.01")) if confluence_scores else None return { "strategy_id": strategy_id, "signals_seen": len(signals), "would_enter": would_enter, "entries_seen": len(entries), "closed_seen": len(closed), "wins": wins, "losses": losses, "net_pnl_usd": str(pnl.quantize(Decimal("0.0001"))), "tv_available_signals": tv_available, "avg_confluence_score": str(avg_conf) if avg_conf is not None else None, "top_blockers": blocker_counts.most_common(15), "blocker_categories": category_counts.most_common(), "top_final_decisions": final_counts.most_common(8), } def load_context_snapshot(runtime_root: Path) -> dict[str, Any]: out: dict[str, Any] = {} confluence_path = runtime_root / "reports" / "market_confluence_latest.json" radar_path = runtime_root / "research" / "coin_opportunity_radar_latest.json" derivatives_path = runtime_root / "market" / "derivatives_latest.json" for key, path in (("confluence", confluence_path), ("radar", radar_path), ("derivatives", derivatives_path)): try: out[key] = json.loads(path.read_text(encoding="utf-8")) except Exception: out[key] = {"status": "not_loaded"} return out def build_report(*, runtime_root: Path = Path("runtime"), experiment_root: Path = Path("runtime/experiments"), limit: int = 5000) -> dict[str, Any]: strategies = {sid: summarize_strategy(sid, runtime_root=experiment_root, limit=limit) for sid in STRATEGIES} context = load_context_snapshot(runtime_root) confluence = context.get("confluence", {}).get("confluence", {}) if isinstance(context.get("confluence"), dict) else {} radar = context.get("radar", {}) if isinstance(context.get("radar"), dict) else {} derivatives = context.get("derivatives", {}) if isinstance(context.get("derivatives"), dict) else {} return { "timestamp": datetime.now(timezone.utc).isoformat(), "schema_version": "confluence_gate_impact.v1", "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False, "market_regime": confluence.get("market_regime"), "eligible_count": confluence.get("eligible_count"), "radar_opportunities": len(radar.get("opportunities") or []), "new_hyperliquid_symbols": radar.get("new_hyperliquid_symbols") or [], "derivatives_source": derivatives.get("source"), "strategies": strategies, } def format_markdown(payload: dict[str, Any]) -> str: lines = [ "# Confluence / MCP Gate Impact Report", "", f"Generated: {payload['timestamp']}", f"Market regime: **{payload.get('market_regime')}**, eligible={payload.get('eligible_count')}", f"Radar opportunities: {payload.get('radar_opportunities')}, new HL symbols: {payload.get('new_hyperliquid_symbols')}", "", "## Strategy Gate Summary", ] for sid, row in payload["strategies"].items(): lines.extend([ "", f"### {sid}", f"- Signals analysed: {row['signals_seen']}", f"- Would-enter signals: {row['would_enter']}", f"- Entries/closed in sampled journal: {row['entries_seen']} / {row['closed_seen']}", f"- Wins/losses/PnL in sampled exits: {row['wins']} / {row['losses']} / {row['net_pnl_usd']} USDC", f"- TV context available: {row['tv_available_signals']}", f"- Avg confluence score: {row['avg_confluence_score']}", "- Top blocker categories: " + ", ".join(f"{k}={v}" for k, v in row["blocker_categories"][:8]), "- Top blockers: " + ", ".join(f"{k}={v}" for k, v in row["top_blockers"][:8]), ]) lines.extend([ "", "## Interpretation", "- If would-enter is near zero while confluence/tradingview blockers dominate, the new data layer is preventing overtrading.", "- Promote toward tiny-live only after new gated Paper/Shadow exits show positive PF, controlled losses, and live preflight is green.", "- Research-only: this report never authorizes live orders.", ]) return "\n".join(lines) + "\n" def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Summarize MCP/confluence gate impact across paper strategy journals.") parser.add_argument("--runtime", default="runtime") parser.add_argument("--experiments", default="runtime/experiments") parser.add_argument("--limit", type=int, default=5000) parser.add_argument("--output", default="runtime/reports/confluence_gate_impact_latest.json") parser.add_argument("--markdown-output", default="runtime/reports/confluence_gate_impact_latest.md") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) payload = build_report(runtime_root=Path(args.runtime), experiment_root=Path(args.experiments), limit=args.limit) output = Path(args.output) md_output = Path(args.markdown_output) output.parent.mkdir(parents=True, exist_ok=True) md_output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") md_output.write_text(format_markdown(payload), encoding="utf-8") if args.json: print(json.dumps(payload, indent=2, sort_keys=True)) else: print(format_markdown(payload)) return 0 if __name__ == "__main__": raise SystemExit(main())