from __future__ import annotations from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_UP from pathlib import Path from typing import Any, Literal import json from src.tools.v76_paper_runtime import STRICT_ID from src.strategies.portfolio_holding_advisor import PORTFOLIO_HOLDING_ID SleeveHorizon = Literal["intraday", "swing", "portfolio"] SleeveMode = Literal["paper_only", "research_only", "advisor_only"] SleeveStatus = Literal["Candidate", "Research", "Advisor", "Blocked"] SWING_RETEST_ID = "candidate_swing_trend_retest_research" @dataclass(frozen=True) class StrategySleeve: sleeve_id: str strategy_id: str name: str horizon: SleeveHorizon mode: SleeveMode purpose: str max_live_authority: str = "none" live_order_allowed: bool = False mainnet_signed_action: bool = False def to_row(self) -> dict[str, str | bool]: return { "sleeve_id": self.sleeve_id, "strategy_id": self.strategy_id, "name": self.name, "horizon": self.horizon, "mode": self.mode, "purpose": self.purpose, "max_live_authority": self.max_live_authority, "live_order_allowed": self.live_order_allowed, "mainnet_signed_action": self.mainnet_signed_action, } SLEEVES: tuple[StrategySleeve, ...] = ( StrategySleeve( sleeve_id="intraday_v76_strict", strategy_id=STRICT_ID, name="Intraday v76 strict", horizon="intraday", mode="paper_only", purpose="Main intraday candidate; confluence-gated lifecycle paper trading only.", ), StrategySleeve( sleeve_id="swing_trend_retest", strategy_id=SWING_RETEST_ID, name="Swing trend/retest", horizon="swing", mode="research_only", purpose="Multi-session trend/retest sleeve placeholder; collect evidence before paper execution.", ), StrategySleeve( sleeve_id="portfolio_holding_advisor", strategy_id=PORTFOLIO_HOLDING_ID, name="Portfolio holding/rebalance advisor", horizon="portfolio", mode="advisor_only", purpose="Allocation/rebalance recommendation sleeve; no exchange execution authority.", ), ) def list_strategy_sleeves() -> list[dict[str, str | bool]]: return [sleeve.to_row() for sleeve in SLEEVES] def _safe_decimal(value: Any) -> Decimal: try: return Decimal(str(value)) except Exception: return Decimal("0") def _fmt_decimal(value: Decimal, places: str = "0.01") -> str: return str(value.quantize(Decimal(places), rounding=ROUND_HALF_UP)) def _journal_metrics(runtime_dir: str | Path, strategy_id: str) -> dict[str, Any]: path = Path(runtime_dir) / "experiments" / strategy_id / "trade_journal.jsonl" entries = 0 exits = 0 pnl = Decimal("0") if not path.exists(): return {"entries": 0, "exits": 0, "net_pnl_usd": Decimal("0")} for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: row = json.loads(line) except Exception: continue if row.get("event") == "entry": entries += 1 if row.get("event") == "exit" and row.get("exit_reason"): exits += 1 pnl += _safe_decimal(row.get("net_pnl_usd") or row.get("realized_pnl_usd")) return {"entries": entries, "exits": exits, "net_pnl_usd": pnl} def build_sleeve_scorecard(*, lifecycle_summary: dict[str, Any], lifecycle_scorecard: dict[str, Any], runtime_dir: str | Path = "runtime") -> dict[str, Any]: """Build a read-only Phase-6 sleeve scorecard. The scorecard separates the three target sleeves without granting live authority. Only the intraday sleeve may currently have lifecycle evidence; swing/portfolio sleeves are explicitly research/advisor-only until their own journals and gates exist. """ cards: list[dict[str, Any]] = [] for sleeve in SLEEVES: blockers: list[str] = [] closed_trades = 0 net_pnl = Decimal("0") winrate = Decimal("0") profit_factor = Decimal("0") status: SleeveStatus evidence_source = "not_started" if sleeve.sleeve_id == "intraday_v76_strict": closed_trades = int(lifecycle_scorecard.get("closed_trades") or 0) net_pnl = _safe_decimal(lifecycle_scorecard.get("net_pnl_usd")) winrate = _safe_decimal(lifecycle_scorecard.get("winrate_pct")) profit_factor = _safe_decimal(lifecycle_scorecard.get("profit_factor")) blockers = [str(item) for item in lifecycle_scorecard.get("blockers") or []] evidence_source = str(lifecycle_scorecard.get("evidence_source") or "true_lifecycle_exits_only") status = "Candidate" if closed_trades >= 30 and not blockers else "Blocked" elif sleeve.sleeve_id == "swing_trend_retest": metrics = _journal_metrics(runtime_dir, sleeve.strategy_id) closed_trades = int(metrics["exits"]) net_pnl = _safe_decimal(metrics["net_pnl_usd"]) evidence_source = "swing_lifecycle_research_journal" if metrics["entries"] else "not_started" status = "Research" blockers = ["research_only_no_live_authority"] if not metrics["entries"]: blockers.append("no_swing_retest_entries_yet") elif closed_trades < 30: blockers.append("swing_sample_too_small") if net_pnl <= 0: blockers.append("swing_net_pnl_not_positive_yet") else: status = "Advisor" blockers = ["advisor_only_no_execution", "portfolio_context_not_connected"] cards.append({ **sleeve.to_row(), "status": status, "evidence_source": evidence_source, "closed_trades": closed_trades, "winrate_pct": _fmt_decimal(winrate), "profit_factor": _fmt_decimal(profit_factor), "net_pnl_usd": _fmt_decimal(net_pnl), "open_positions": int(lifecycle_summary.get("open_positions") or 0) if sleeve.sleeve_id == "intraday_v76_strict" else 0, "blockers": blockers, "live_allowed": False, }) return { "version": "strategy_sleeves.v1", "runtime_dir": str(runtime_dir), "sleeves": cards, "live_allowed": False, "mainnet_signed_action": False, } def format_sleeve_scorecard(scorecard: dict[str, Any]) -> str: sleeves = scorecard.get("sleeves") if isinstance(scorecard.get("sleeves"), list) else [] parts = [] for row in sleeves: raw_blockers = row.get("blockers") if isinstance(row, dict) else [] blockers = raw_blockers if isinstance(raw_blockers, list) else [] blocker_text = ",".join(str(item) for item in blockers[:3]) if blockers else "keine" parts.append( f"{row.get('sleeve_id')}={row.get('status')} " f"mode={row.get('mode')} closed={row.get('closed_trades')} " f"pf={row.get('profit_factor')} pnl={row.get('net_pnl_usd')} blocker={blocker_text}" ) return f"Strategy Sleeves v1: {' | '.join(parts) if parts else 'keine'}, live=nein"