from __future__ import annotations from collections import Counter from dataclasses import dataclass from decimal import Decimal, InvalidOperation import json from pathlib import Path from typing import Any from src.ctb_copy.models import CopyCostInput, CopyGateConfig, PositionCopySignal, SignalDecision from src.ctb_copy.read_only_guard import DEFAULT_READ_ONLY_GUARD, ReadOnlyGuard from src.ctb_copy.risk.copy_pretrade_gate import evaluate_copy_signal from src.ctb_copy.runtime_paths import CopyResearchPaths, day_string, utc_now_iso from src.ctb_copy.shadow.follower_fill_model import estimate_follower_net_pnl from src.ctb_copy.shadow.position_journal import append_position_decisions, append_shadow_portfolio, write_shadow_result from src.ctb_copy.shadow.shadow_models import ( Side, ShadowCopyDecision, ShadowCostEstimate, ShadowDecisionType, ShadowDeltaType, ShadowFollowerPosition, ShadowLeaderPosition, ShadowPortfolioState, ShadowPositionDelta, ShadowRunSummary, ) REASON_MAP = { "fill_level_copying_disabled": "BLOCK_FILL_LEVEL_COPYING_DISABLED", "position_age_below_6h": "BLOCK_POSITION_TOO_YOUNG", "leader_already_above_0_5r_profit": "BLOCK_LEADER_TOO_FAR_IN_PROFIT", "distance_to_leader_entry_too_large": "BLOCK_ENTRY_DISTANCE_TOO_LARGE", "spread_too_wide": "BLOCK_SPREAD_TOO_WIDE", "market_depth_not_ok": "BLOCK_DEPTH_TOO_THIN", "leader_liquidation_distance_too_close": "BLOCK_LIQUIDATION_DISTANCE_TOO_SMALL", "funding_extreme": "BLOCK_FUNDING_EXTREME", "duplicate_exposure_already_present": "BLOCK_DUPLICATE_EXPOSURE", } CANDIDATE_DELTAS = {ShadowDeltaType.NEW_POSITION, ShadowDeltaType.INCREASE_POSITION} DEFAULT_MIN_TOP_DEPTH_USD = Decimal("10000") DEFAULT_NOTIONAL_USD = Decimal("100") DEFAULT_SLIPPAGE_PCT = Decimal("0.03") DEFAULT_DELAY_PENALTY_PCT = Decimal("0.02") def D(value: Any, default: str = "0") -> Decimal: try: if value is None or value == "": return Decimal(default) return Decimal(str(value)) except (InvalidOperation, ValueError): return Decimal(default) def signed_size(position: ShadowLeaderPosition) -> Decimal: if position.side == "short": return -abs(position.size) if position.side == "long": return abs(position.size) return Decimal("0") def side_from_signed_size(size: Decimal) -> Side: if size > 0: return "long" if size < 0: return "short" return "flat" def classify_delta(previous_size: Decimal, current_size: Decimal) -> ShadowDeltaType: if previous_size == current_size: return ShadowDeltaType.UNCHANGED_POSITION if previous_size == 0 and current_size != 0: return ShadowDeltaType.NEW_POSITION if previous_size != 0 and current_size == 0: return ShadowDeltaType.CLOSE_POSITION if (previous_size > 0 > current_size) or (previous_size < 0 < current_size): return ShadowDeltaType.FLIP_POSITION if abs(current_size) > abs(previous_size): return ShadowDeltaType.INCREASE_POSITION return ShadowDeltaType.DECREASE_POSITION def compute_position_delta(previous: ShadowLeaderPosition | None, current: ShadowLeaderPosition) -> ShadowPositionDelta: previous_signed = signed_size(previous) if previous else Decimal("0") current_signed = signed_size(current) delta_type = classify_delta(previous_signed, current_signed) return ShadowPositionDelta( leader_id=current.leader_id, leader_type=current.leader_type, symbol=current.symbol, side=current.side, previous_size=previous_signed, current_size=current_signed, size_delta=current_signed - previous_signed, delta_type=delta_type, leader_entry_price=current.leader_entry_price, current_mid=current.current_mid, observed_at=current.observed_at, position_age_hours=current.position_age_hours, leader_unrealized_pnl_pct=current.leader_unrealized_pnl_pct, leader_liquidation_distance_pct=current.leader_liquidation_distance_pct, funding_rate=current.funding_rate, spread_pct=current.spread_pct, top_depth_usd=current.top_depth_usd, source_snapshot_ids=tuple(x for x in (previous.source_snapshot_id if previous else None, current.source_snapshot_id) if x), ) def compute_deltas(previous_positions: list[ShadowLeaderPosition], current_positions: list[ShadowLeaderPosition]) -> list[ShadowPositionDelta]: prev_by_key = {(p.leader_id, p.symbol): p for p in previous_positions} current_by_key = {(p.leader_id, p.symbol): p for p in current_positions} deltas = [compute_position_delta(prev_by_key.get(key), current) for key, current in sorted(current_by_key.items())] for key, previous in sorted(prev_by_key.items()): if key not in current_by_key: closed = ShadowLeaderPosition( leader_id=previous.leader_id, leader_type=previous.leader_type, symbol=previous.symbol, side="flat", size=Decimal("0"), leader_entry_price=previous.leader_entry_price, current_mid=previous.current_mid, observed_at=utc_now_iso(), position_age_hours=previous.position_age_hours, leader_unrealized_pnl_pct=previous.leader_unrealized_pnl_pct, leader_liquidation_distance_pct=previous.leader_liquidation_distance_pct, funding_rate=previous.funding_rate, spread_pct=previous.spread_pct, top_depth_usd=previous.top_depth_usd, source_snapshot_id=previous.source_snapshot_id, ) deltas.append(compute_position_delta(previous, closed)) return deltas def _leader_profit_r(delta: ShadowPositionDelta) -> Decimal: # v0 approximation: treat 1R as 1 pct leader move. Conservative and testable until stop-distance metadata exists. return delta.leader_unrealized_pnl_pct def _distance_to_entry_pct(delta: ShadowPositionDelta) -> Decimal: if delta.leader_entry_price == 0: return Decimal("999") return abs((delta.current_mid - delta.leader_entry_price) / delta.leader_entry_price * Decimal("100")) def estimate_entry_costs(delta: ShadowPositionDelta, *, slippage_pct: Decimal = DEFAULT_SLIPPAGE_PCT, delay_penalty_pct: Decimal = DEFAULT_DELAY_PENALTY_PCT) -> ShadowCostEstimate: item = CopyCostInput( notional_usd=DEFAULT_NOTIONAL_USD, gross_pnl_usd=Decimal("0"), spread_pct=delta.spread_pct, modeled_slippage_pct=slippage_pct, delay_penalty_usd=DEFAULT_NOTIONAL_USD / Decimal("100") * delay_penalty_pct, ) costs = estimate_follower_net_pnl(item) entry_fee_pct = item.entry_fee_pct exit_fee_pct = item.exit_fee_pct entry_total = entry_fee_pct + item.spread_pct + item.modeled_slippage_pct + delay_penalty_pct roundtrip = entry_total + exit_fee_pct # Ensure formula result is actually exercised, not decorative. _ = costs.net_pnl_usd return ShadowCostEstimate( entry_fee_pct=entry_fee_pct, exit_fee_pct=exit_fee_pct, spread_cost_pct=item.spread_pct, modeled_slippage_pct=item.modeled_slippage_pct, delay_penalty_pct=delay_penalty_pct, estimated_total_entry_cost_pct=entry_total, estimated_roundtrip_cost_pct=roundtrip, ) def simulated_entry_price(delta: ShadowPositionDelta, costs: ShadowCostEstimate) -> Decimal: adjustment = (delta.spread_pct + costs.modeled_slippage_pct + costs.delay_penalty_pct) / Decimal("100") if delta.side == "short": return delta.current_mid * (Decimal("1") - adjustment) return delta.current_mid * (Decimal("1") + adjustment) @dataclass(frozen=True) class ShadowEngineConfig: min_top_depth_usd: Decimal = DEFAULT_MIN_TOP_DEPTH_USD duplicate_symbols: tuple[str, ...] = () allow_swing_override: bool = False class ShadowPositionEngine: def __init__(self, guard: ReadOnlyGuard = DEFAULT_READ_ONLY_GUARD, config: ShadowEngineConfig | None = None) -> None: self.guard = guard self.config = config or ShadowEngineConfig() def evaluate_delta(self, delta: ShadowPositionDelta) -> ShadowCopyDecision: costs = estimate_entry_costs(delta) guard_reasons: list[str] = [] try: self.guard.assert_safe() except RuntimeError: guard_reasons.append("BLOCK_READ_ONLY_GUARD_FAILED") if delta.delta_type not in CANDIDATE_DELTAS: return self._decision(delta, ShadowDecisionType.IGNORED, (f"IGNORE_{delta.delta_type.value.upper()}",), costs, None) age = delta.position_age_hours if self.config.allow_swing_override and age == 0: age = CopyGateConfig().min_position_age_hours signal = PositionCopySignal( leader_id=delta.leader_id, coin=delta.symbol, side=delta.side, position_age_hours=age, leader_profit_r=_leader_profit_r(delta), distance_to_leader_entry_pct=_distance_to_entry_pct(delta), spread_pct=delta.spread_pct, depth_ok=delta.top_depth_usd >= self.config.min_top_depth_usd, leader_liquidation_distance_pct=delta.leader_liquidation_distance_pct, funding_rate_hourly_pct=delta.funding_rate, duplicate_exposure=delta.symbol in self.config.duplicate_symbols, is_position_level=True, ) gate = evaluate_copy_signal(signal) reasons = tuple(REASON_MAP.get(reason, reason) for reason in gate.reasons if reason != "ok_for_shadow_only") + tuple(guard_reasons) decision_type = ShadowDecisionType.ALLOWED if gate.decision == SignalDecision.ALLOW_SHADOW and not guard_reasons else ShadowDecisionType.BLOCKED if decision_type == ShadowDecisionType.ALLOWED: reasons = ("ALLOW_SHADOW_POSITION",) entry = simulated_entry_price(delta, costs) else: entry = None return self._decision(delta, decision_type, reasons, costs, entry) def evaluate(self, deltas: list[ShadowPositionDelta]) -> list[ShadowCopyDecision]: return [self.evaluate_delta(delta) for delta in deltas] def _decision(self, delta: ShadowPositionDelta, decision: ShadowDecisionType, reasons: tuple[str, ...], costs: ShadowCostEstimate, entry: Decimal | None) -> ShadowCopyDecision: return ShadowCopyDecision( leader_id=delta.leader_id, leader_type=delta.leader_type, symbol=delta.symbol, side=delta.side, previous_size=delta.previous_size, current_size=delta.current_size, size_delta=delta.size_delta, delta_type=delta.delta_type, decision=decision, reason_codes=reasons, leader_entry_price=delta.leader_entry_price, current_mid=delta.current_mid, observed_at=delta.observed_at, position_age_hours=delta.position_age_hours, leader_unrealized_pnl_pct=delta.leader_unrealized_pnl_pct, leader_liquidation_distance_pct=delta.leader_liquidation_distance_pct, funding_rate=delta.funding_rate, spread_pct=delta.spread_pct, top_depth_usd=delta.top_depth_usd, follower_sim_entry_price=entry, estimated_roundtrip_cost_pct=costs.estimated_roundtrip_cost_pct, estimated_slippage_pct=costs.modeled_slippage_pct, estimated_delay_penalty_pct=costs.delay_penalty_pct, estimated_total_entry_cost_pct=costs.estimated_total_entry_cost_pct, risk_unit_R=Decimal("1"), estimated_costs=costs, source_snapshot_ids=delta.source_snapshot_ids, read_only_guard=self.guard.as_dict(), ) def build_shadow_portfolio(decisions: list[ShadowCopyDecision], observed_at: str | None = None, guard: ReadOnlyGuard = DEFAULT_READ_ONLY_GUARD) -> ShadowPortfolioState: positions: list[ShadowFollowerPosition] = [] for decision in decisions: if decision.decision != ShadowDecisionType.ALLOWED or decision.follower_sim_entry_price is None: continue notional = DEFAULT_NOTIONAL_USD simulated_size = notional / decision.follower_sim_entry_price if decision.follower_sim_entry_price else Decimal("0") positions.append( ShadowFollowerPosition( leader_id=decision.leader_id, leader_type=decision.leader_type, symbol=decision.symbol, side=decision.side, simulated_size=simulated_size, follower_sim_entry_price=decision.follower_sim_entry_price, notional_usd=notional, risk_unit_R=decision.risk_unit_R, opened_at=decision.observed_at, source_decision=decision, ) ) return ShadowPortfolioState( observed_at=observed_at or utc_now_iso(), positions=tuple(positions), total_notional_usd=sum((p.notional_usd for p in positions), Decimal("0")), read_only_guard=guard.as_dict(), ) def summarize_shadow_run(day: str, decisions: list[ShadowCopyDecision], paths: CopyResearchPaths, status: str | None = None) -> ShadowRunSummary: counts = Counter(decision.decision.value for decision in decisions) reason_counts: Counter[str] = Counter() for decision in decisions: if decision.decision == ShadowDecisionType.BLOCKED: reason_counts.update(decision.reason_codes) estimated_notional = DEFAULT_NOTIONAL_USD * Decimal(counts.get("allowed", 0)) resolved_status = status or ("ok_no_position_deltas" if not decisions else "ok") return ShadowRunSummary( status=resolved_status, day=day, decisions_count=len(decisions), allowed_count=counts.get("allowed", 0), blocked_count=counts.get("blocked", 0), ignored_count=counts.get("ignored", 0), top_block_reasons=dict(reason_counts.most_common(10)), symbols_observed=tuple(sorted({decision.symbol for decision in decisions})), leaders_observed=tuple(sorted({decision.leader_id for decision in decisions})), estimated_total_notional_allowed=estimated_notional, decision_journal_path=str(paths.shadow_position_decisions_file(day)), portfolio_journal_path=str(paths.shadow_portfolio_file(day)), result_path=str(paths.shadow_result_file(day)), read_only_guard=DEFAULT_READ_ONLY_GUARD.as_dict(), ) def run_shadow_engine_once(previous_positions: list[ShadowLeaderPosition], current_positions: list[ShadowLeaderPosition], paths: CopyResearchPaths, day: str | None = None) -> ShadowRunSummary: resolved_day = day or day_string() deltas = compute_deltas(previous_positions, current_positions) engine = ShadowPositionEngine() decisions = engine.evaluate(deltas) portfolio = build_shadow_portfolio(decisions) if decisions: append_position_decisions(decisions, paths.shadow_position_decisions_file(resolved_day)) append_shadow_portfolio(portfolio, paths.shadow_portfolio_file(resolved_day)) summary = summarize_shadow_run(resolved_day, decisions, paths) write_shadow_result(summary, paths.shadow_result_file(resolved_day)) return summary def _extract_asset_positions(snapshot_row: dict[str, Any]) -> list[ShadowLeaderPosition]: leader_id = str(snapshot_row.get("wallet_address") or snapshot_row.get("leader_id") or "unknown") observed_at = str(snapshot_row.get("observed_at_ms") or snapshot_row.get("observed_at") or utc_now_iso()) source_snapshot_id = str(snapshot_row.get("source_snapshot_id") or snapshot_row.get("observed_at_ms") or "") or None raw = snapshot_row.get("raw", {}) if isinstance(snapshot_row.get("raw", {}), dict) else {} clearing = raw.get("clearinghouseState", {}) if isinstance(raw.get("clearinghouseState", {}), dict) else {} positions = clearing.get("assetPositions", []) if isinstance(clearing.get("assetPositions", []), list) else [] result: list[ShadowLeaderPosition] = [] for row in positions: pos = row.get("position", row) if isinstance(row, dict) else {} if not isinstance(pos, dict): continue symbol = str(pos.get("coin") or pos.get("symbol") or "") size = D(pos.get("szi") or pos.get("size") or pos.get("positionSize")) if not symbol or size == 0: continue entry = D(pos.get("entryPx") or pos.get("entry_price") or pos.get("entryPrice"), "0") current_mid = D(pos.get("midPx") or pos.get("markPx") or pos.get("current_mid") or entry, "0") result.append( ShadowLeaderPosition( leader_id=leader_id, leader_type="wallet", symbol=symbol, side=side_from_signed_size(size), size=abs(size), leader_entry_price=entry, current_mid=current_mid, observed_at=observed_at, position_age_hours=D(pos.get("position_age_hours"), "0"), leader_unrealized_pnl_pct=D(pos.get("leader_unrealized_pnl_pct") or pos.get("returnOnEquity"), "0"), leader_liquidation_distance_pct=D(pos.get("leader_liquidation_distance_pct"), "999"), funding_rate=D(pos.get("funding_rate"), "0"), spread_pct=D(pos.get("spread_pct"), "0.03"), top_depth_usd=D(pos.get("top_depth_usd"), "999999999"), source_snapshot_id=source_snapshot_id, ) ) return result def load_wallet_positions_from_snapshot_file(path: Path) -> list[ShadowLeaderPosition]: if not path.exists(): return [] rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] if not rows: return [] return _extract_asset_positions(rows[-1])