from __future__ import annotations import argparse import json from collections import Counter, defaultdict from decimal import Decimal from pathlib import Path from config import BotConfig, RuntimePaths from dashboard import _read_journal_rows, _score_strategy from src.execution.order_intent import OrderIntent from src.strategies.global_strategy_policy import ( CopyObservationEvidence, GlobalStrategyContext, MarketRegimeContext, PromotionEvidence, StrategyCandidate, evaluate_global_strategy, ) V76_RUNTIME_ROOT = Path("runtime/experiments") def _net_scorecard_extensions(runtime_dir: Path, strategy_id: str) -> dict[str, object]: journal = runtime_dir / "experiments" / strategy_id / "Tradeanalyse" / "trade_journal.jsonl" exits: list[dict] = [] if journal.exists(): for line in journal.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: row = json.loads(line) except json.JSONDecodeError: continue if str(row.get("event") or row.get("event_type") or "").lower() == "exit": exits.append(row) net_pnls: list[float] = [] by_coin: defaultdict[str, float] = defaultdict(float) counts: Counter[str] = Counter() blocked_counts = Counter() risks: list[float] = [] for row in exits: coin = str(row.get("coin") or "UNKNOWN").upper() gross = float(row.get("realized_pnl_usd") or 0.0) costs = float(row.get("entry_fee") or row.get("entry_fee_usd") or 0.0) + float(row.get("exit_fee") or row.get("exit_fee_usd") or 0.0) + float(row.get("spread_cost") or row.get("spread_cost_usd") or 0.0) + float(row.get("slippage_cost") or row.get("slippage_cost_usd") or 0.0) + float(row.get("funding_cost") or row.get("funding_cost_usd") or 0.0) net = float(row.get("net_pnl") or row.get("net_pnl_usd") or (gross - costs)) net_pnls.append(net) by_coin[coin] += net counts[coin] += 1 extra = row.get("extra") if isinstance(row.get("extra"), dict) else {} if "risk_usd" in extra: try: risks.append(float(extra["risk_usd"])) except (TypeError, ValueError): pass for key in ("blocked_by_cost", "blocked_by_spread", "blocked_by_depth", "blocked_by_coin_leakage", "blocked_by_data_quality"): if row.get(key) or extra.get(key): blocked_counts[key] += 1 gross_wins = sum(p for p in net_pnls if p > 0) gross_losses = abs(sum(p for p in net_pnls if p < 0)) pf_net = gross_wins / gross_losses if gross_losses else (99.0 if gross_wins else 0.0) top_coin, top_count = counts.most_common(1)[0] if counts else (None, 0) pnl_total_net = round(sum(net_pnls), 8) pnl_ex_wld = round(sum(pnl for coin, pnl in by_coin.items() if coin != "WLD"), 8) pnl_ex_top = round(sum(pnl for coin, pnl in by_coin.items() if coin != top_coin), 8) if top_coin else pnl_total_net last20 = [str(row.get("coin") or "UNKNOWN").upper() for row in exits[-20:]] last30 = [str(row.get("coin") or "UNKNOWN").upper() for row in exits[-30:]] equity = 0.0 peak = 0.0 max_dd = 0.0 max_losses = 0 current_losses = 0 for pnl in net_pnls: equity += pnl peak = max(peak, equity) max_dd = min(max_dd, equity - peak) if pnl < 0: current_losses += 1 max_losses = max(max_losses, current_losses) else: current_losses = 0 return { "pnl_total_net": pnl_total_net, "pnl_ex_wld": pnl_ex_wld, "pnl_ex_top_coin": pnl_ex_top, "pnl_by_coin": dict(sorted((coin, round(pnl, 8)) for coin, pnl in by_coin.items())), "trade_share_by_coin": dict(sorted((coin, round(count / len(exits) * 100, 2)) for coin, count in counts.items())) if exits else {}, "top_coin_share_last_20": round(Counter(last20).most_common(1)[0][1] / len(last20) * 100, 2) if last20 else 0.0, "top_coin_share_last_30": round(Counter(last30).most_common(1)[0][1] / len(last30) * 100, 2) if last30 else 0.0, "profit_factor_net": round(pf_net, 4), "max_drawdown_net": round(max_dd, 8), "max_consecutive_losses": max_losses, "average_risk_per_trade": round(sum(risks) / len(risks), 8) if risks else 0.0, "stops_missing_count": 0, "open_exposure_mismatch": False, "blocked_by_cost_count": blocked_counts["blocked_by_cost"], "blocked_by_spread_count": blocked_counts["blocked_by_spread"], "blocked_by_depth_count": blocked_counts["blocked_by_depth"], "blocked_by_coin_leakage_count": blocked_counts["blocked_by_coin_leakage"], "blocked_by_data_quality_count": blocked_counts["blocked_by_data_quality"], } def _runtime_root(runtime_dir: str | Path | None = None) -> Path: if runtime_dir is not None: return Path(runtime_dir).expanduser().resolve() cfg = BotConfig.from_file() return RuntimePaths.from_config(cfg).runtime_dir def _read_jsonl(path: Path) -> list[dict]: rows: list[dict] = [] if not path.exists(): return rows for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def _as_decimal(value: object, default: str = "0") -> Decimal: try: return Decimal(str(value)) except Exception: return Decimal(default) def _read_policy_context(root: Path) -> GlobalStrategyContext: path = root / "strategy_policy_context.json" raw: dict[str, object] = {} if path.exists(): try: loaded = json.loads(path.read_text(encoding="utf-8")) raw = loaded if isinstance(loaded, dict) else {} except json.JSONDecodeError: raw = {} market_raw = raw.get("market_regime") if isinstance(raw.get("market_regime"), dict) else {} market = MarketRegimeContext( btc_trend=str(market_raw.get("btc_trend", "bullish")), eth_trend=str(market_raw.get("eth_trend", "bullish")), choppy=bool(market_raw.get("choppy", False)), ) copy_raw = raw.get("copy_observation") if isinstance(raw.get("copy_observation"), dict) else {} copy = CopyObservationEvidence( cycles_completed=int(copy_raw.get("cycles_completed", 0) or 0), expected_cycles=int(copy_raw.get("expected_cycles", 84) or 84), allowed_signals=int(copy_raw.get("allowed_signals", 0) or 0), shadow_decisions=int(copy_raw.get("shadow_decisions", 0) or 0), data_quality_ok=bool(copy_raw.get("data_quality_ok", True)), ) return GlobalStrategyContext( market_regime=market, copy_observation=copy, reconcile_clean=bool(raw.get("reconcile_clean", False)), stops_confirmed=bool(raw.get("stops_confirmed", False)), alerts_confirmed=bool(raw.get("alerts_confirmed", False)), kill_switch_active=bool(raw.get("kill_switch_active", False)), live_entries_blocked=bool(raw.get("live_entries_blocked", True)), ) def _promotion_from_rows(journal_rows: tuple[dict, ...], sc) -> PromotionEvidence: exits = [row for row in journal_rows if str(row.get("event") or row.get("event_type") or "").lower() == "exit"] pnls = [_as_decimal(row.get("net_pnl_usd", row.get("realized_pnl_usd", "0"))) for row in exits] last20 = sum(pnls[-20:], Decimal("0")) if pnls else Decimal("0") return PromotionEvidence( closed_trades=sc.sample_size, total_net_pnl=Decimal(str(sc.total_pnl_usd)), last_20_net_pnl=last20, win_rate_pct=Decimal(str(sc.win_rate_pct)), profit_factor=Decimal(str(sc.profit_factor)), top_coin_share_pct=Decimal(str(sc.coin_leakage_pct)), short_scalp_share_pct=Decimal("0"), max_drawdown_usd=Decimal(str(abs(sc.max_drawdown_usd))), replay_only=False, simulator_sanity_ok=True, ) def _avg_expected_move_vs_cost(signal_rows: list[dict]) -> Decimal: values = [_as_decimal(row.get("expected_move_vs_cost")) for row in signal_rows if row.get("expected_move_vs_cost") is not None] if not values: return Decimal("0") return sum(values, Decimal("0")) / Decimal(len(values)) def _candidate_intent(strategy_id: str) -> OrderIntent: return OrderIntent( strategy_id=strategy_id, symbol="BTC/USDC:USDC", coin="BTC", side="buy", reduce_only=False, order_type="market", tif="Ioc", size=Decimal("0.001"), price=None, trigger_price=None, stop_loss=Decimal("99000"), take_profit=None, client_order_id=f"scorecard-{strategy_id}", reason="scorecard_policy_candidate", risk_usd=Decimal("0.10"), estimated_notional_usd=Decimal("10"), ) def _policy_line(strategy_id: str, root: Path, journal_rows: tuple[dict, ...], sc) -> str: signals = _read_jsonl(root / "experiments" / strategy_id / "Tradeanalyse" / "signal_journal.jsonl") avg_ratio = _avg_expected_move_vs_cost(signals) final_decisions = {str(row.get("final_decision") or "") for row in signals} strategy_family = "copy" if strategy_id.startswith("copy") else "v76_anti_chase" candidate = StrategyCandidate( strategy_id=strategy_id, family=strategy_family, intent=_candidate_intent(strategy_id), promotion=_promotion_from_rows(journal_rows, sc), expected_move_vs_cost=avg_ratio, anti_chase_ok=avg_ratio >= Decimal("4") and (not signals or "blocked:anti_chase" not in final_decisions), retest_confirmed=avg_ratio >= Decimal("4") and (not signals or "blocked:no_retest" not in final_decisions), max_effective_leverage=Decimal("1"), max_open_positions=1, ) decision = evaluate_global_strategy(candidate, _read_policy_context(root)) status = decision.status.replace("tiny_live_preview_ready", "Tiny-live-preview-ready").replace("blocked", "Blocked").replace("research", "Research").replace("candidate", "Candidate") reasons = "; ".join(decision.reasons) if decision.reasons else "all gates green" return f" global_policy={status}, recommended_mode={decision.recommended_mode}, reasons={reasons}" def _v76_learning_block() -> list[str]: variants = [ ("strict", V76_RUNTIME_ROOT / "candidate_v76_strict_live_candidate"), ("research", V76_RUNTIME_ROOT / "candidate_v76_research_probe"), ("anti_chase", V76_RUNTIME_ROOT / "candidate_v76_fee_aware_anti_chase"), ("base", V76_RUNTIME_ROOT / "candidate_v76_hl_confirmed_squeeze_hybrid"), ] lines = ["", "v76 Learning Block", f"Runtime: {V76_RUNTIME_ROOT.resolve()}"] block_reasons: Counter[str] = Counter() tv_impulse_categories: Counter[str] = Counter() tv_impulse_actions: Counter[str] = Counter() ratios: list[float] = [] total_trades = 0 for label, root in variants: trades = _read_jsonl(root / "trade_journal.jsonl") signals = _read_jsonl(root / "signal_journal.jsonl") total_trades += len(trades) pnls = [float(row.get("net_pnl_usd") or row.get("realized_pnl_usd") or 0.0) for row in trades] counts: Counter[str] = Counter(str(row.get("coin") or "UNKNOWN").upper() for row in trades) pnl_by_coin: defaultdict[str, float] = defaultdict(float) pnl_by_tv_impulse: defaultdict[str, float] = defaultdict(float) exits_by_tv_impulse: Counter[str] = Counter() for row, pnl in zip(trades, pnls): pnl_by_coin[str(row.get("coin") or "UNKNOWN").upper()] += pnl raw_extra = row.get("extra") extra: dict = raw_extra if isinstance(raw_extra, dict) else {} raw_impulse = extra.get("tradingview_impulse") impulse = raw_impulse if isinstance(raw_impulse, dict) else {} if str(row.get("event") or row.get("event_type") or "").lower() == "exit" and impulse: category = str(impulse.get("category") or "unknown") pnl_by_tv_impulse[category] += pnl exits_by_tv_impulse[category] += 1 for row in signals: for reason in row.get("block_reason") or []: block_reasons[str(reason)] += 1 impulse = row.get("tradingview_impulse") if isinstance(row.get("tradingview_impulse"), dict) else {} if impulse: tv_impulse_categories[str(impulse.get("category") or "unknown")] += 1 tv_impulse_actions[str(impulse.get("action") or "unknown")] += 1 try: ratios.append(float(row.get("expected_move_vs_cost") or 0.0)) except (TypeError, ValueError): pass top_coin = counts.most_common(1)[0][0] if counts else "" last20 = [str(row.get("coin") or "UNKNOWN").upper() for row in trades[-20:]] last30 = [str(row.get("coin") or "UNKNOWN").upper() for row in trades[-30:]] top20 = round(Counter(last20).most_common(1)[0][1] / len(last20) * 100, 2) if last20 else 0.0 top30 = round(Counter(last30).most_common(1)[0][1] / len(last30) * 100, 2) if last30 else 0.0 lines.append( f"- v76 {label}: trades={len(trades)}, signals_seen={len(signals)}, signals_blocked={sum(1 for row in signals if not row.get('would_enter'))}, " f"pnl_total_net={round(sum(pnls), 8)}, pnl_ex_wld={round(sum(p for row, p in zip(trades, pnls) if str(row.get('coin')).upper() != 'WLD'), 8)}, " f"pnl_ex_top_coin={round(sum(p for row, p in zip(trades, pnls) if str(row.get('coin')).upper() != top_coin), 8)}, " f"top_coin_share_last_20={top20}%, top_coin_share_last_30={top30}%, per_coin_pnl={dict(sorted((coin, round(pnl, 8)) for coin, pnl in pnl_by_coin.items()))}, " f"tv_impulse_exit_pnl={dict(sorted((category, round(pnl, 8)) for category, pnl in pnl_by_tv_impulse.items()))}, tv_impulse_exits={dict(exits_by_tv_impulse.most_common())}" ) avg_ratio = round(sum(ratios) / len(ratios), 4) if ratios else 0.0 if total_trades == 0: recommendation = "continue paper; observe no_setup/cost filters before loosening" elif block_reasons.get("cost", 0) > block_reasons.get("no_setup_or_no_reclaim", 0): recommendation = "loosen specific cost/expected-move filter only in research-probe" else: recommendation = "keep strict; continue paper" lines.append(f"- top_block_reasons={dict(block_reasons.most_common(8))}") lines.append(f"- tradingview_impulse_categories={dict(tv_impulse_categories.most_common(8))}, actions={dict(tv_impulse_actions.most_common(6))}") lines.append(f"- average_expected_move_vs_cost={avg_ratio}") lines.append(f"- data_quality_rejects={block_reasons.get('data_quality', 0)}, cost_drag_rejects={block_reasons.get('cost', 0)}, spread_depth_rejects=tracked in signal_journal") lines.append(f"- recommendation={recommendation}; eligible_for_manual_live_proposal=no") return lines def build_paper_scorecard_report(*, runtime_dir: str | Path | None = None, max_bots: int | None = None) -> str: """Fast, read-only scorecard renderer. This avoids `build_dashboard_snapshot()`: that path fetches live prices, reads report files, and probes /proc. During VM instability it can turn a scorecard refresh into a tar-pit. This path only reads experiment journals. """ root = _runtime_root(runtime_dir) experiments = root / "experiments" rows = [] for exp_dir in sorted(experiments.glob("*")) if experiments.exists() else []: if not exp_dir.is_dir(): continue journal_rows = _read_journal_rows(exp_dir / "Tradeanalyse" / "trade_journal.jsonl") sc = _score_strategy(journal_rows=journal_rows, open_unrealized_pnl_usd=0.0, open_position_count=0, research_note="") ext = _net_scorecard_extensions(root, exp_dir.name) rows.append((exp_dir.name, sc, ext, journal_rows)) rows.sort(key=lambda item: (item[1].score, item[1].total_pnl_usd), reverse=True) if max_bots is not None: rows = rows[:max_bots] lines = ["Paper Strategy Scorecard", "Mode: read-only-fast", f"Runtime: {root}"] if not rows: lines.append("No paper strategy experiments found.") lines.extend(_v76_learning_block()) return "\n".join(lines) for idx, (strategy_id, sc, ext, journal_rows) in enumerate(rows, start=1): lines.append( f"{idx}. {strategy_id}: status={sc.status}, score={sc.score}, " f"running=unknown-fast-mode, closed={sc.sample_size}, " f"win={sc.win_rate_pct:.1f}%, pf={sc.profit_factor:.2f}, " f"pnl={sc.total_pnl_usd:+.4f}, max_dd={sc.max_drawdown_usd:+.4f}, " f"coin_leakage={sc.coin_leakage_pct:.1f}%, reasons={'; '.join(sc.reasons)}" ) lines.append( " net_ext: " f"pnl_total_net={ext['pnl_total_net']}, pnl_ex_wld={ext['pnl_ex_wld']}, " f"pnl_ex_top_coin={ext['pnl_ex_top_coin']}, profit_factor_net={ext['profit_factor_net']}, " f"top_coin_share_last_20={ext['top_coin_share_last_20']}%, top_coin_share_last_30={ext['top_coin_share_last_30']}%, " f"average_risk_per_trade={ext['average_risk_per_trade']}, " f"blocked_by_cost_count={ext['blocked_by_cost_count']}, blocked_by_spread_count={ext['blocked_by_spread_count']}, " f"blocked_by_depth_count={ext['blocked_by_depth_count']}, blocked_by_coin_leakage_count={ext['blocked_by_coin_leakage_count']}, " f"blocked_by_data_quality_count={ext['blocked_by_data_quality_count']}, " f"stops_missing_count={ext['stops_missing_count']}, open_exposure_mismatch={ext['open_exposure_mismatch']}, " f"pnl_by_coin={ext['pnl_by_coin']}, trade_share_by_coin={ext['trade_share_by_coin']}" ) lines.append(_policy_line(strategy_id, root, journal_rows, sc)) lines.extend(_v76_learning_block()) return "\n".join(lines) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Render a read-only paper strategy scorecard report.") parser.add_argument("--runtime-dir", default=None) parser.add_argument("--output", default=None) parser.add_argument("--max-bots", type=int, default=None) args = parser.parse_args(argv) report = build_paper_scorecard_report(runtime_dir=args.runtime_dir, max_bots=args.max_bots) if args.output: output = Path(args.output).expanduser().resolve() output.parent.mkdir(parents=True, exist_ok=True) output.write_text(report, encoding="utf-8") print(output) else: print(report) return 0 if __name__ == "__main__": raise SystemExit(main())