from __future__ import annotations import argparse import json import time from collections import defaultdict from datetime import datetime from decimal import Decimal from pathlib import Path from typing import Any from src.strategies.trend_retest_anti_chase_v2 import STRATEGY_VERSION HEALTH_MAX_AGE_SECONDS = 180 MAX_PROMOTION_DRAWDOWN_USD = Decimal("2") MAX_PROMOTION_LOSS_STREAK = 3 def _d(value: Any, default: str = "0") -> Decimal: try: return Decimal(str(value)) except Exception: return Decimal(default) def _profit_factor(values: list[Decimal]) -> Decimal: gross_profit = sum((value for value in values if value > 0), Decimal("0")) gross_loss = -sum((value for value in values if value < 0), Decimal("0")) return gross_profit / gross_loss if gross_loss > 0 else (Decimal("999") if gross_profit > 0 else Decimal("0")) def _load_jsonl(path: Path) -> tuple[list[dict[str, Any]], int]: rows: list[dict[str, Any]] = [] invalid = 0 if not path.exists(): return rows, invalid for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: row = json.loads(line) except json.JSONDecodeError: invalid += 1 continue if isinstance(row, dict): rows.append(row) else: invalid += 1 return rows, invalid def _load_state(path: Path) -> tuple[dict[str, Any], bool]: if not path.exists(): return {}, False try: payload = json.loads(path.read_text(encoding="utf-8")) except Exception: return {}, False return (payload, True) if isinstance(payload, dict) else ({}, False) def _lifecycle_key(row: dict[str, Any]) -> tuple[str, str, str]: return ( str(row.get("run_id") or ""), str(row.get("data_window_id") or ""), str(row.get("coin") or "").upper(), ) def _risk_path(values: list[Decimal]) -> tuple[Decimal, int]: equity = Decimal("0") peak = Decimal("0") max_drawdown = Decimal("0") loss_streak = 0 max_loss_streak = 0 for value in values: equity += value peak = max(peak, equity) max_drawdown = max(max_drawdown, peak - equity) if value < 0: loss_streak += 1 max_loss_streak = max(max_loss_streak, loss_streak) else: loss_streak = 0 return max_drawdown, max_loss_streak def _health_integrity(runtime: Path, *, strategy_id: str, strategy_version: str) -> tuple[dict[str, Any], list[str]]: rows, invalid = _load_jsonl(runtime / "runtime_health.jsonl") blockers: list[str] = [] if invalid: blockers.append("runtime_health_invalid_jsonl") latest = rows[-1] if rows else {} age = float("inf") if latest: try: stamp = datetime.fromisoformat(str(latest.get("timestamp")).replace("Z", "+00:00")) age = max(0.0, time.time() - stamp.timestamp()) except Exception: blockers.append("runtime_health_timestamp_invalid") else: blockers.append("runtime_health_missing") if age > HEALTH_MAX_AGE_SECONDS: blockers.append("runtime_health_stale") if latest and latest.get("status") != "ok": blockers.append("runtime_health_not_ok") if latest and latest.get("strategy_id") != strategy_id: blockers.append("runtime_health_strategy_id_mismatch") if latest and latest.get("strategy_version") != strategy_version: blockers.append("runtime_health_strategy_version_mismatch") if latest and ( latest.get("paper_trading") is not True or latest.get("live_order_allowed") is not False or latest.get("mainnet_signed_action") is not False ): blockers.append("runtime_health_safety_flags_invalid") return {"age_seconds": None if age == float("inf") else round(age, 3), "latest_status": latest.get("status"), "invalid_rows": invalid}, blockers def build_report(runtime_dir: str | Path, *, strategy_version: str = STRATEGY_VERSION) -> dict[str, Any]: runtime = Path(runtime_dir) strategy_id = runtime.name rows, invalid_trade_rows = _load_jsonl(runtime / "trade_journal.jsonl") version_rows = [row for row in rows if row.get("strategy_version") == strategy_version] foreign_identity_rows = sum(1 for row in version_rows if row.get("strategy_id") != strategy_id) entries = [row for row in version_rows if row.get("event") == "entry"] exits = [row for row in version_rows if row.get("event") == "exit"] entry_by_key: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) exit_by_key: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) invalid_identity_rows = 0 for row in entries: key = _lifecycle_key(row) if not all(key): invalid_identity_rows += 1 else: entry_by_key[key].append(row) for row in exits: key = _lifecycle_key(row) if not all(key): invalid_identity_rows += 1 else: exit_by_key[key].append(row) duplicate_entries = sum(max(0, len(group) - 1) for group in entry_by_key.values()) duplicate_exits = sum(max(0, len(group) - 1) for group in exit_by_key.values()) orphan_entries = sorted(key for key in entry_by_key if key not in exit_by_key) orphan_exits = sorted(key for key in exit_by_key if key not in entry_by_key) paired_keys = sorted(key for key in entry_by_key if len(entry_by_key[key]) == 1 and len(exit_by_key.get(key, [])) == 1) paired_entries = [entry_by_key[key][0] for key in paired_keys] paired_exits = [exit_by_key[key][0] for key in paired_keys] paired_exits.sort(key=lambda row: str(row.get("timestamp") or "")) pnl = [_d(row.get("net_pnl_usd")) for row in paired_exits] last30 = pnl[-30:] cost_rows = [row for row in paired_exits if row.get("gross_pnl_usd") is not None] effective_costs = sum((_d(row.get("gross_pnl_usd")) - _d(row.get("net_pnl_usd")) for row in cost_rows), Decimal("0")) winner_cost_drags: list[Decimal] = [] for row in cost_rows: gross = _d(row.get("gross_pnl_usd")) net = _d(row.get("net_pnl_usd")) if gross > 0 and net > 0: winner_cost_drags.append((gross - net) / gross * Decimal("100")) high_cost_drag_winners = sum(1 for value in winner_cost_drags if value > Decimal("50")) high_cost_drag_share = Decimal(high_cost_drag_winners) / Decimal(len(winner_cost_drags)) * Decimal("100") if winner_cost_drags else Decimal("0") gross_profit_before_costs = sum((_d(row.get("gross_pnl_usd")) for row in cost_rows if _d(row.get("gross_pnl_usd")) > 0), Decimal("0")) aggregate_cost_drag = effective_costs / gross_profit_before_costs * Decimal("100") if gross_profit_before_costs > 0 else Decimal("0") by_coin: dict[str, list[Decimal]] = defaultdict(list) by_week: dict[str, list[Decimal]] = defaultdict(list) for row, value in zip(paired_exits, pnl): coin = str(row.get("coin") or "UNKNOWN").upper() by_coin[coin].append(value) try: stamp = datetime.fromisoformat(str(row.get("timestamp")).replace("Z", "+00:00")) year, week, _ = stamp.isocalendar() by_week[f"{year}-W{week:02d}"].append(value) except Exception: by_week["unknown"].append(value) per_coin = {coin: sum(values, Decimal("0")) for coin, values in sorted(by_coin.items())} positive_total = sum((value for value in per_coin.values() if value > 0), Decimal("0")) top_coin_share = max((value / positive_total * Decimal("100") for value in per_coin.values() if value > 0), default=Decimal("0")) if positive_total > 0 else Decimal("0") leave_one_out: dict[str, dict[str, str]] = {} for coin in by_coin: remaining = [value for other, values in by_coin.items() if other != coin for value in values] leave_one_out[coin] = {"net_pnl": str(sum(remaining, Decimal("0"))), "profit_factor": str(_profit_factor(remaining))} profitable_weeks = sum(1 for week, values in by_week.items() if week != "unknown" and sum(values, Decimal("0")) > 0) state, state_valid = _load_state(runtime / "state.json") raw_open_positions = state.get("open_positions") if state_valid else None open_positions_container_valid = raw_open_positions is None or isinstance(raw_open_positions, dict) open_positions: dict[str, Any] = raw_open_positions if isinstance(raw_open_positions, dict) else {} raw_pending_orders = state.get("pending_orders") if state_valid else None pending_orders_container_valid = raw_pending_orders is None or isinstance(raw_pending_orders, dict) pending_orders: dict[str, Any] = raw_pending_orders if isinstance(raw_pending_orders, dict) else {} state_version_ok = state_valid and state.get("strategy_version") == strategy_version proxy_rows = sum(1 for row in paired_entries + paired_exits if row.get("proxy_inputs_used") is not False) source_attribution_covered = sum(1 for row in paired_entries if isinstance(row.get("source_attribution"), list) and bool(row.get("source_attribution"))) source_attribution_required = strategy_version.startswith("v78.2") stop_covered = sum(1 for row in paired_entries if _d(row.get("stop_loss")) > 0) unique_windows = len({str(row.get("data_window_id")) for row in paired_entries if row.get("data_window_id")}) pf = _profit_factor(pnl) recent_pf = _profit_factor(last30) max_drawdown, max_loss_streak = _risk_path(pnl) leave_one_out_ok = bool(leave_one_out) and all(_d(row["net_pnl"]) > 0 and _d(row["profit_factor"]) >= Decimal("1.10") for row in leave_one_out.values()) health, health_blockers = _health_integrity(runtime, strategy_id=strategy_id, strategy_version=strategy_version) integrity = { "trade_jsonl_valid": invalid_trade_rows == 0, "strategy_identity_valid": foreign_identity_rows == 0 and invalid_identity_rows == 0, "one_entry_one_final_exit": duplicate_entries == 0 and duplicate_exits == 0 and not orphan_entries and not orphan_exits, "state_valid": state_valid, "lifecycle_state_containers_valid": open_positions_container_valid and pending_orders_container_valid, "state_strategy_version_matches": state_version_ok, "runtime_health_valid_and_fresh": not health_blockers, } gates = { **integrity, "fresh_lifecycle_exits_50": len(paired_exits) >= 50, "three_profitable_weeks": profitable_weeks >= 3, "overall_profit_factor_1_25": pf >= Decimal("1.25"), "last30_profit_factor_1_25": len(last30) >= 30 and recent_pf >= Decimal("1.25"), "net_pnl_positive": sum(pnl, Decimal("0")) > 0, "last30_net_pnl_positive": len(last30) >= 30 and sum(last30, Decimal("0")) > 0, "leave_one_coin_out_positive": leave_one_out_ok, "top_coin_share_max_35pct": top_coin_share <= Decimal("35"), "unique_data_windows": unique_windows == len(paired_entries), "no_proxy_inputs": proxy_rows == 0, "source_attribution_coverage_100pct": not source_attribution_required or (len(paired_entries) > 0 and source_attribution_covered == len(paired_entries)), "entry_stop_coverage_100pct": len(paired_entries) > 0 and stop_covered == len(paired_entries), "cost_efficiency_coverage_100pct": len(paired_exits) > 0 and len(cost_rows) == len(paired_exits), "high_cost_drag_winner_share_max_20pct": bool(winner_cost_drags) and high_cost_drag_share <= Decimal("20"), "max_drawdown_usd_2": max_drawdown <= MAX_PROMOTION_DRAWDOWN_USD, "max_consecutive_losses_3": max_loss_streak <= MAX_PROMOTION_LOSS_STREAK, "no_open_positions_for_promotion": not open_positions and not pending_orders, } blockers = [key for key, allowed in gates.items() if not allowed] blockers.extend(item for item in health_blockers if item not in blockers) return { "schema_version": "v77_promotion_report.v2", "strategy_id": strategy_id, "strategy_version": strategy_version, "runtime_dir": str(runtime), "entries": len(entries), "closed_lifecycles": len(paired_exits), "unique_data_windows": unique_windows, "profitable_weeks": profitable_weeks, "net_pnl_usd": str(sum(pnl, Decimal("0"))), "profit_factor": str(pf), "last30_net_pnl_usd": str(sum(last30, Decimal("0"))), "last30_profit_factor": str(recent_pf), "max_drawdown_usd": str(max_drawdown), "max_consecutive_losses": max_loss_streak, "top_coin_share_pct": str(top_coin_share), "per_coin_net_pnl": {coin: str(value) for coin, value in per_coin.items()}, "leave_one_coin_out": leave_one_out, "proxy_rows": proxy_rows, "source_attribution_coverage": f"{source_attribution_covered}/{len(paired_entries)}", "stop_coverage": f"{stop_covered}/{len(paired_entries)}", "cost_efficiency": { "coverage": f"{len(cost_rows)}/{len(paired_exits)}", "gross_profit_before_costs_usd": str(gross_profit_before_costs), "effective_costs_usd": str(effective_costs), "aggregate_cost_drag_pct_of_gross_profit": str(aggregate_cost_drag), "winner_count": len(winner_cost_drags), "high_cost_drag_winners": high_cost_drag_winners, "high_cost_drag_winner_share_pct": str(high_cost_drag_share), "high_cost_drag_threshold_pct": "50", }, "open_positions": len(open_positions), "pending_orders": len(pending_orders), "integrity": { "invalid_trade_rows": invalid_trade_rows, "foreign_identity_rows": foreign_identity_rows, "invalid_identity_rows": invalid_identity_rows, "duplicate_entries": duplicate_entries, "duplicate_exits": duplicate_exits, "orphan_entries": len(orphan_entries), "orphan_exits": len(orphan_exits), "state_valid": state_valid, "open_positions_container_valid": open_positions_container_valid, "pending_orders_container_valid": pending_orders_container_valid, "health": health, }, "gates": gates, "blockers": list(dict.fromkeys(blockers)), "promotion_eligible": not blockers, "recommendation": "eligible_for_manual_paper_evidence_review" if not blockers else "continue_paper_only", "paper_trading": True, "execution_allowed": False, "live_order_allowed": False, "mainnet_signed_action": False, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Build integrity-checked net-of-cost paper promotion evidence; never enables execution.") parser.add_argument("--runtime-dir", required=True) parser.add_argument("--strategy-version", default=STRATEGY_VERSION) parser.add_argument("--output") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) report = build_report(args.runtime_dir, strategy_version=args.strategy_version) output = Path(args.output) if args.output else Path(args.runtime_dir) / "promotion_report_latest.json" output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8") print(json.dumps(report, indent=2, sort_keys=True) if args.json else f"paper promotion: eligible={report['promotion_eligible']} blockers={','.join(report['blockers']) or 'none'}") return 0 if __name__ == "__main__": raise SystemExit(main())