#!/usr/bin/env python3 """VIOLET base-fraction sizing study. Read-only analysis against recorded CH trade data. Produces: - a machine-readable JSON report under prod/VIOLET_dev/reports/ - a short markdown findings note alongside it The study is scoped by prod/docs/VIOLET_STUDY_SPEC__BASE_FRACTION_SIZING.md. It does not modify production code or write to production tables. """ from __future__ import annotations import argparse import bisect import csv import dataclasses import json import math import statistics import sys from collections import defaultdict from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Iterable, Sequence import numpy as np PROJECT_ROOT = Path("/mnt/dolphinng5_predict") REPORTS_DIR = PROJECT_ROOT / "prod" / "VIOLET_dev" / "reports" CH_URL = "http://localhost:8123/" CH_USER = "dolphin" CH_KEY = "dolphin_ch_2026" BASE_FRACTION_F0 = 0.20 TRANSLATOR_CAP = 3.0 BASE_GRID = np.array( [0.20, 0.22, 0.24, 0.25, 0.26, 0.28, 0.30, 0.32, 0.333, 0.34, 0.36, 0.38, 0.40, 0.45, 0.50], dtype=np.float64, ) @dataclass(frozen=True) class TradeRow: trade_id: str ts: datetime asset: str side: str entry_price: float exit_price: float quantity: float pnl: float pnl_pct: float exit_reason: str leverage: float capital_before: float capital_after: float bars_held: int regime_signal: int vel_div_entry: float boost_at_entry: float beta_at_entry: float posture: str our_leverage: float composite_hash: int | None = None scalar_hash: int | None = None regime: str | None = None fingerprint_confidence: float | None = None fingerprint_vel_div: float | None = None fingerprint_dvol: float | None = None notional_quote: float | None = None fill_quality_score: float | None = None fill_quality_class: str | None = None fill_rows: int = 0 taker_fill_rows: int = 0 maker_fill_rows: int = 0 @dataclass(frozen=True) class FingerprintRow: ts: datetime regime: str composite_hash: int scalar_hash: int confidence: float raw_vel_div: float raw_dvol: float final_score: float def _query_tsv(sql: str) -> list[dict[str, str]]: import urllib.request req = urllib.request.Request( CH_URL, data=sql.encode(), headers={"X-ClickHouse-User": CH_USER, "X-ClickHouse-Key": CH_KEY}, ) with urllib.request.urlopen(req, timeout=120) as resp: text = resp.read().decode() lines = [line for line in text.splitlines() if line.strip()] if not lines: return [] reader = csv.DictReader(lines, delimiter="\t") return list(reader) def _fmt_ts(dt: datetime) -> str: if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(timezone.utc) return dt.strftime("%Y-%m-%d %H:%M:%S.%f") def _parse_dt(s: str) -> datetime: if isinstance(s, datetime): return s s = s.strip() if s.endswith("Z"): s = s[:-1] + "+00:00" try: return datetime.fromisoformat(s).astimezone(timezone.utc) except ValueError: # ClickHouse DateTime64 TSV may arrive without timezone suffix. return datetime.strptime(s, "%Y-%m-%d %H:%M:%S.%f").replace(tzinfo=timezone.utc) def _parse_float(v: str | None, default: float = 0.0) -> float: if v is None or v == "" or v == "\\N": return float(default) return float(v) def _parse_int(v: str | None, default: int = 0) -> int: if v is None or v == "" or v == "\\N": return int(default) return int(float(v)) def load_clean_trades() -> list[TradeRow]: sql = """ WITH dedup AS ( SELECT trade_id, max(event_ts) AS ts, argMax(asset, event_ts) AS asset, argMax(side, event_ts) AS side, argMax(entry_price, event_ts) AS entry_price, argMax(exit_price, event_ts) AS exit_price, argMax(quantity, event_ts) AS quantity, argMax(pnl, event_ts) AS pnl, argMax(pnl_pct, event_ts) AS pnl_pct, argMax(exit_reason, event_ts) AS exit_reason, argMax(leverage, event_ts) AS leverage, argMax(capital_before, event_ts) AS capital_before, argMax(capital_after, event_ts) AS capital_after, argMax(bars_held, event_ts) AS bars_held, argMax(regime_signal, event_ts) AS regime_signal, argMax(vel_div_entry, event_ts) AS vel_div_entry, argMax(boost_at_entry, event_ts) AS boost_at_entry, argMax(beta_at_entry, event_ts) AS beta_at_entry, argMax(posture, event_ts) AS posture, argMax(our_leverage, event_ts) AS our_leverage FROM ( SELECT trade_id, ts AS event_ts, asset, side, entry_price, exit_price, quantity, pnl, pnl_pct, exit_reason, leverage, capital_before, capital_after, bars_held, regime_signal, vel_div_entry, boost_at_entry, beta_at_entry, posture, our_leverage FROM dolphin.trade_events ) GROUP BY trade_id ) SELECT trade_id, ts, asset, side, entry_price, exit_price, quantity, pnl, pnl_pct, exit_reason, leverage, capital_before, capital_after, bars_held, regime_signal, vel_div_entry, boost_at_entry, beta_at_entry, posture, our_leverage FROM dedup WHERE exit_reason != 'HIBERNATE_HALT' AND bars_held > 0 ORDER BY ts FORMAT TSVWithNames """ rows = _query_tsv(sql) out: list[TradeRow] = [] for row in rows: out.append( TradeRow( trade_id=row["trade_id"], ts=_parse_dt(row["ts"]), asset=row["asset"], side=row["side"], entry_price=_parse_float(row["entry_price"]), exit_price=_parse_float(row["exit_price"]), quantity=_parse_float(row["quantity"]), pnl=_parse_float(row["pnl"]), pnl_pct=_parse_float(row["pnl_pct"]), exit_reason=row["exit_reason"], leverage=_parse_float(row["leverage"]), capital_before=_parse_float(row["capital_before"]), capital_after=_parse_float(row["capital_after"]), bars_held=_parse_int(row["bars_held"]), regime_signal=_parse_int(row["regime_signal"]), vel_div_entry=_parse_float(row["vel_div_entry"]), boost_at_entry=_parse_float(row["boost_at_entry"], 1.0), beta_at_entry=_parse_float(row["beta_at_entry"], 1.0), posture=row["posture"], our_leverage=_parse_float(row["our_leverage"]), ) ) return out def load_exec_quality() -> dict[str, dict[str, Any]]: sql = """ SELECT trade_id, maxIf(notional_quote, record_kind = 'trade_summary') AS notional_quote, maxIf(fill_quality_score, record_kind = 'trade_summary') AS fill_quality_score, anyIf(fill_quality_class, record_kind = 'trade_summary') AS fill_quality_class, countIf(record_kind = 'fill') AS fill_rows, countIf(record_kind = 'fill' AND liquidity_side = 'TAKER') AS taker_fill_rows, countIf(record_kind = 'fill' AND liquidity_side = 'MAKER') AS maker_fill_rows FROM dolphin.trade_execution_quality GROUP BY trade_id FORMAT TSVWithNames """ rows = _query_tsv(sql) out: dict[str, dict[str, Any]] = {} for row in rows: out[row["trade_id"]] = { "notional_quote": _parse_float(row["notional_quote"]), "fill_quality_score": _parse_float(row["fill_quality_score"]), "fill_quality_class": row["fill_quality_class"], "fill_rows": _parse_int(row["fill_rows"]), "taker_fill_rows": _parse_int(row["taker_fill_rows"]), "maker_fill_rows": _parse_int(row["maker_fill_rows"]), } return out def load_fingerprints(start_ts: datetime, end_ts: datetime) -> list[FingerprintRow]: sql = f""" SELECT ts, regime, composite_hash, scalar_hash, confidence, raw_vel_div, raw_dvol, final_score FROM dolphin.maras_fingerprint WHERE ts >= toDateTime64('{_fmt_ts(start_ts)}', 6, 'UTC') AND ts <= toDateTime64('{_fmt_ts(end_ts)}', 6, 'UTC') ORDER BY ts FORMAT TSVWithNames """ rows = _query_tsv(sql) out: list[FingerprintRow] = [] for row in rows: out.append( FingerprintRow( ts=_parse_dt(row["ts"]), regime=row["regime"], composite_hash=_parse_int(row["composite_hash"]), scalar_hash=_parse_int(row["scalar_hash"]), confidence=_parse_float(row["confidence"]), raw_vel_div=_parse_float(row["raw_vel_div"]), raw_dvol=_parse_float(row["raw_dvol"]), final_score=_parse_float(row["final_score"]), ) ) return out def attach_fingerprint(trades: list[TradeRow], fps: list[FingerprintRow]) -> list[TradeRow]: fp_ts = [fp.ts for fp in fps] out: list[TradeRow] = [] for trade in trades: idx = bisect.bisect_right(fp_ts, trade.ts) - 1 if idx >= 0: fp = fps[idx] trade = dataclasses.replace( trade, composite_hash=fp.composite_hash, scalar_hash=fp.scalar_hash, regime=fp.regime, fingerprint_confidence=fp.confidence, fingerprint_vel_div=fp.raw_vel_div, fingerprint_dvol=fp.raw_dvol, ) out.append(trade) return out def build_trade_set() -> tuple[list[TradeRow], dict[str, Any]]: trades = load_clean_trades() if not trades: raise RuntimeError("no clean trades returned from ClickHouse") eq = load_exec_quality() enriched: list[TradeRow] = [] for trade in trades: meta = eq.get(trade.trade_id, {}) notional = _parse_float(str(meta.get("notional_quote", 0.0)), 0.0) if notional <= 0: notional = abs(trade.entry_price * trade.quantity) enriched.append( dataclasses.replace( trade, notional_quote=notional, fill_quality_score=meta.get("fill_quality_score"), fill_quality_class=meta.get("fill_quality_class"), fill_rows=int(meta.get("fill_rows", 0)), taker_fill_rows=int(meta.get("taker_fill_rows", 0)), maker_fill_rows=int(meta.get("maker_fill_rows", 0)), ) ) fps = load_fingerprints(trades[0].ts - timedelta(hours=1), trades[-1].ts) enriched = attach_fingerprint(enriched, fps) return enriched, { "trade_rows": len(trades), "exec_quality_rows": len(eq), "fingerprint_rows": len(fps), } def slippage_bps_model(notional: float, median_notional: float, *, taker_only: bool = True) -> float: """Conservative parametric impact proxy. Direct slippage telemetry in trade_execution_quality is mostly null for this dataset, so we assume a taker-heavy execution floor and a sublinear size-dependent impact term. """ base = 0.35 if taker_only else 0.20 impact = 0.15 * math.sqrt(max(notional, 1.0) / max(median_notional, 1.0)) return base + impact def replay_equity( trades: Sequence[TradeRow], fraction: float, *, slippage_enabled: bool = True, median_notional: float, ruin_threshold: float = 0.50, ) -> dict[str, Any]: if not trades: raise ValueError("no trades") start_capital = trades[0].capital_before if trades[0].capital_before > 0 else 69_000.0 capital = float(start_capital) equity_points: list[tuple[datetime, float]] = [] daily_close: dict[str, float] = {} daily_peak: dict[str, float] = {} clipped = 0 total = 0 returns: list[float] = [] daily_equity: dict[str, float] = {} for trade in trades: total += 1 leverage_eff = min(trade.leverage, TRANSLATOR_CAP / max(fraction, 1e-12)) if trade.leverage * fraction > TRANSLATOR_CAP: clipped += 1 notional = capital * fraction * leverage_eff slip_bps = slippage_bps_model( notional, median_notional, taker_only=(trade.taker_fill_rows > 0 or trade.fill_rows > 0), ) if slippage_enabled else 0.0 trade_return = fraction * leverage_eff * (trade.pnl_pct - slip_bps / 10_000.0) capital *= 1.0 + trade_return returns.append(trade_return) equity_points.append((trade.ts, capital)) day = trade.ts.date().isoformat() daily_equity[day] = capital # Fill daily series from first to last trade day. first_day = trades[0].ts.date() last_day = trades[-1].ts.date() day = first_day last_equity = start_capital daily_series: list[tuple[str, float]] = [] while day <= last_day: key = day.isoformat() if key in daily_equity: last_equity = daily_equity[key] daily_series.append((key, last_equity)) day += timedelta(days=1) daily_returns = [] prev = start_capital for _, eq in daily_series: daily_returns.append((eq / prev) - 1.0 if prev else 0.0) prev = eq peak = start_capital max_dd = 0.0 underwater_start: str | None = None longest_underwater_days = 0.0 current_underwater_days = 0.0 prev_day: str | None = None for day_str, eq in daily_series: if eq >= peak: if underwater_start is not None and prev_day is not None: start = datetime.fromisoformat(underwater_start).date() end = datetime.fromisoformat(prev_day).date() longest_underwater_days = max( longest_underwater_days, (end - start).days + 1 ) peak = eq underwater_start = None current_underwater_days = 0.0 else: if underwater_start is None: underwater_start = day_str current_underwater_days += 1.0 max_dd = max(max_dd, 1.0 - eq / peak if peak > 0 else 0.0) prev_day = day_str if underwater_start is not None and prev_day is not None: start = datetime.fromisoformat(underwater_start).date() end = datetime.fromisoformat(prev_day).date() longest_underwater_days = max(longest_underwater_days, (end - start).days + 1) years = max((trades[-1].ts - trades[0].ts).total_seconds() / (365.25 * 24 * 3600), 1.0 / 365.25) if capital > 0 and start_capital > 0: log_growth = math.log(capital / start_capital) annual_log = log_growth / years if annual_log > 700.0: cagr = float("inf") elif annual_log < -700.0: cagr = -1.0 else: cagr = math.expm1(annual_log) else: cagr = -1.0 ann_sharpe, ann_sortino, downside_dev = _daily_risk_metrics(daily_returns) ruin_prob = _bootstrap_ruin_prob( trades=trades, fraction=fraction, median_notional=median_notional, slippage_enabled=slippage_enabled, ruin_threshold=ruin_threshold, ) return { "fraction": fraction, "start_capital": start_capital, "final_capital": capital, "cagr": cagr, "max_drawdown": max_dd, "calmar": (cagr / max_dd) if max_dd > 0 else float("inf"), "sharpe": ann_sharpe, "sortino": ann_sortino, "downside_deviation": downside_dev, "ruin_prob": ruin_prob, "pct_trades_clipped_at_3x": (clipped / total * 100.0) if total else 0.0, "longest_underwater_days": longest_underwater_days, "n_trades": total, "daily_returns": daily_returns, "equity_points": equity_points, "returns": returns, } def _daily_risk_metrics(daily_returns: Sequence[float]) -> tuple[float, float, float]: if len(daily_returns) < 2: return 0.0, 0.0, 0.0 arr = np.asarray(daily_returns, dtype=np.float64) mean = float(np.mean(arr)) std = float(np.std(arr, ddof=1)) if len(arr) > 1 else 0.0 downside = arr[arr < 0.0] downside_dev = float(np.std(downside, ddof=1)) if len(downside) > 1 else float(np.std(np.minimum(arr, 0.0), ddof=0)) sharpe = (mean / std * math.sqrt(365.25)) if std > 0 else 0.0 sortino = (mean / downside_dev * math.sqrt(365.25)) if downside_dev > 0 else 0.0 return sharpe, sortino, downside_dev def _bootstrap_ruin_prob( *, trades: Sequence[TradeRow], fraction: float, median_notional: float, slippage_enabled: bool, ruin_threshold: float, n_boot: int = 256, seed: int = 17, ) -> float: rng = np.random.default_rng(seed) n = len(trades) if n == 0: return 0.0 leverage = np.array([min(t.leverage, TRANSLATOR_CAP / max(fraction, 1e-12)) for t in trades], dtype=np.float64) pnl_pct = np.array([t.pnl_pct for t in trades], dtype=np.float64) taker_only = np.array([(t.taker_fill_rows > 0 or t.fill_rows > 0) for t in trades], dtype=np.float64) ruin = 0 for _ in range(n_boot): idx = rng.integers(0, n, size=n) capital = 1.0 floor = ruin_threshold for j in idx: lev = leverage[j] notional = capital * fraction * lev slip = slippage_bps_model(notional, median_notional, taker_only=bool(taker_only[j])) if slippage_enabled else 0.0 r = fraction * lev * (pnl_pct[j] - slip / 10_000.0) capital *= 1.0 + r if capital <= floor: ruin += 1 break return ruin / n_boot def cap_binding_curve(trades: Sequence[TradeRow], fractions: Sequence[float]) -> list[dict[str, float]]: out = [] n = len(trades) for f in fractions: clipped = sum(1 for t in trades if t.leverage * f > TRANSLATOR_CAP) out.append({"fraction": float(f), "pct_clipped": (clipped / n * 100.0) if n else 0.0}) return out def _group_by_hash(trades: Sequence[TradeRow]) -> dict[int, list[TradeRow]]: groups: dict[int, list[TradeRow]] = defaultdict(list) for trade in trades: if trade.composite_hash is None: continue groups[int(trade.composite_hash)].append(trade) for arr in groups.values(): arr.sort(key=lambda t: t.ts) return groups def _bucket_stats(trades: Sequence[TradeRow], fraction: float, median_notional: float) -> dict[str, Any]: groups = _group_by_hash(trades) bucket_results = [] for h, rows in groups.items(): if len(rows) < 8: continue rep = replay_equity(rows, fraction, slippage_enabled=True, median_notional=median_notional) bucket_results.append( { "composite_hash": int(h), "n_trades": len(rows), "final_capital": rep["final_capital"], "max_drawdown": rep["max_drawdown"], "cagr": rep["cagr"], "calmar": rep["calmar"], "ruin_prob": rep["ruin_prob"], } ) bucket_results.sort(key=lambda x: (x["max_drawdown"], -x["n_trades"]), reverse=True) return { "bucket_results": bucket_results, "worst_bucket": bucket_results[0] if bucket_results else None, } def slippage_model_summary(trades: Sequence[TradeRow]) -> dict[str, Any]: notionals = np.array([t.notional_quote or abs(t.entry_price * t.quantity) for t in trades], dtype=np.float64) if len(notionals) == 0: median_notional = 1.0 else: median_notional = float(np.median(notionals)) taker_fill_rate = float( sum(1 for t in trades if t.taker_fill_rows > 0 or t.fill_rows > 0) / max(len(trades), 1) ) observed_direct = sum( 1 for t in trades if math.isfinite(float(t.fill_quality_score or 0.0)) and (t.fill_quality_score or 0.0) not in (0.0, None) ) return { "type": "conservative_parametric_proxy", "observed_direct_slippage_rows": 0, "observed_direct_slippage_trade_rows": observed_direct, "taker_fill_rate": taker_fill_rate, "median_notional_quote": median_notional, "formula": "slippage_bps = 0.35 + 0.15 * sqrt(notional / median_notional)", "assumption": "execution-quality rows do not expose non-null slippage_bps; all fill rows are taker, so this is a conservative proxy", } def analyze() -> dict[str, Any]: trades, counts = build_trade_set() median_notional = float(np.median([t.notional_quote or abs(t.entry_price * t.quantity) for t in trades])) fractions = np.unique(np.concatenate([BASE_GRID, np.round(np.arange(0.20, 0.501, 0.01), 3)])).astype(np.float64) slippage_summary = slippage_model_summary(trades) results_no_slip = [] results_slip = [] for f in fractions: results_no_slip.append(replay_equity(trades, float(f), slippage_enabled=False, median_notional=median_notional)) results_slip.append(replay_equity(trades, float(f), slippage_enabled=True, median_notional=median_notional)) best_slip = max(results_slip, key=lambda x: (x["calmar"], x["final_capital"])) best_no_slip = max(results_no_slip, key=lambda x: (x["calmar"], x["final_capital"])) refined = np.unique( np.concatenate([ fractions, np.round(np.arange(max(0.20, best_slip["fraction"] - 0.03), min(0.50, best_slip["fraction"] + 0.03) + 0.0001, 0.005), 3), ]) ).astype(np.float64) if len(refined) > len(fractions): results_slip = [replay_equity(trades, float(f), slippage_enabled=True, median_notional=median_notional) for f in refined] results_no_slip = [replay_equity(trades, float(f), slippage_enabled=False, median_notional=median_notional) for f in refined] fractions = refined best_slip = max(results_slip, key=lambda x: (x["calmar"], x["final_capital"])) best_no_slip = max(results_no_slip, key=lambda x: (x["calmar"], x["final_capital"])) cap_curve = cap_binding_curve(trades, fractions) bucket_summary = _bucket_stats(trades, float(best_slip["fraction"]), median_notional) cap_by_fraction = {round(float(row["fraction"]), 3): float(row["pct_clipped"]) for row in cap_curve} feasible_rows = [r for r in results_slip if cap_by_fraction.get(round(float(r["fraction"]), 3), 0.0) == 0.0] practical_best = max(feasible_rows, key=lambda x: (x["calmar"], x["final_capital"])) if feasible_rows else best_slip # Kelly anchor: use unit-fraction returns (return at f=1.0, ignoring cap/slip) unit_returns = np.array([t.pnl_pct * min(t.leverage, TRANSLATOR_CAP / 1.0) for t in trades], dtype=np.float64) kelly_grid = np.linspace(0.01, 0.50, 200) kelly_log_growth = [] for f in kelly_grid: growth = np.mean(np.log1p(np.clip(f * unit_returns, -0.95, None))) kelly_log_growth.append(float(growth)) kelly_idx = int(np.argmax(kelly_log_growth)) kelly_fraction = float(kelly_grid[kelly_idx]) fractional_kelly = float(min(best_slip["fraction"], max(0.25 * kelly_fraction, 0.20))) # Stress scenario: worst-hash bucket loss multiplied slightly and injected once. stress = None if bucket_summary["worst_bucket"] is not None: worst_hash = bucket_summary["worst_bucket"]["composite_hash"] worst_rows = sorted([t for t in trades if t.composite_hash == worst_hash], key=lambda t: t.ts) if worst_rows: worst_trade = min(worst_rows, key=lambda t: t.pnl_pct) stress_trade = dataclasses.replace(worst_trade, pnl_pct=min(-0.01, worst_trade.pnl_pct * 1.5)) stress_rows = list(worst_rows) + [stress_trade] stress_rows.sort(key=lambda t: t.ts) stress = { "worst_hash": int(worst_hash), "worst_trade_id": worst_trade.trade_id, "stress_pnl_pct": float(stress_trade.pnl_pct), "per_fraction": [ { "fraction": float(f), "final_capital": replay_equity(stress_rows, float(f), slippage_enabled=True, median_notional=median_notional)["final_capital"], "ruin_prob": replay_equity(stress_rows, float(f), slippage_enabled=True, median_notional=median_notional)["ruin_prob"], } for f in fractions ], } recommendation = { "recommended_fraction": float(practical_best["fraction"]), "recommended_basis": "best_calmar_among_zero_clip_fractions", "recommended_final_capital": float(practical_best["final_capital"]), "recommended_cagr": float(practical_best["cagr"]), "recommended_max_drawdown": float(practical_best["max_drawdown"]), "recommended_calmar": float(practical_best["calmar"]), "recommended_ruin_prob": float(practical_best["ruin_prob"]), "recommended_clip_pct": float(cap_by_fraction.get(round(float(practical_best["fraction"]), 3), 0.0)), "alt_fraction_floor": float(fractional_kelly), "no_slippage_best_fraction": float(best_no_slip["fraction"]), "slippage_best_fraction": float(best_slip["fraction"]), "optimizer_fraction": float(best_slip["fraction"]), "optimizer_final_capital": float(best_slip["final_capital"]), "optimizer_cagr": float(best_slip["cagr"]), "optimizer_max_drawdown": float(best_slip["max_drawdown"]), "optimizer_calmar": float(best_slip["calmar"]), "optimizer_ruin_prob": float(best_slip["ruin_prob"]), "optimizer_clip_pct": float(next(c for c in cap_curve if abs(c["fraction"] - best_slip["fraction"]) < 1e-9)["pct_clipped"]), "delta_final_capital_vs_base_0p20": float( next(r for r in results_slip if abs(r["fraction"] - 0.20) < 1e-9)["final_capital"] ), } return { "generated_at": datetime.now(timezone.utc).isoformat(), "study_spec": "prod/docs/VIOLET_STUDY_SPEC__BASE_FRACTION_SIZING.md", "source_counts": counts, "clean_trade_count": len(trades), "clean_trade_window": { "start": trades[0].ts.isoformat(), "end": trades[-1].ts.isoformat(), }, "slippage_model": slippage_summary, "cap_curve": cap_curve, "results": { "no_slippage": results_no_slip, "slippage_adjusted": results_slip, }, "kelly": { "kelly_fraction": kelly_fraction, "fractional_kelly_anchor": fractional_kelly, "kelly_log_growth_grid": [{"fraction": float(f), "log_growth": float(g)} for f, g in zip(kelly_grid, kelly_log_growth)], }, "bucket_summary": bucket_summary, "stress_scenario": stress, "recommendation": recommendation, } def _best_row(rows: Sequence[dict[str, Any]]) -> dict[str, Any]: return max(rows, key=lambda x: (x["calmar"], x["final_capital"])) def _format_pct(x: float) -> str: return f"{x * 100.0:.2f}%" def write_report(report: dict[str, Any]) -> tuple[Path, Path]: REPORTS_DIR.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") json_path = REPORTS_DIR / f"base_fraction_study_{ts}.json" md_path = REPORTS_DIR / f"base_fraction_study_{ts}.md" json_path.write_text(json.dumps(report, indent=2, sort_keys=True, default=str)) best = report["recommendation"] base = next(r for r in report["results"]["slippage_adjusted"] if abs(r["fraction"] - 0.20) < 1e-9) recommended_cap = next(c for c in report["cap_curve"] if abs(c["fraction"] - best["recommended_fraction"]) < 1e-9) optimizer_cap = next(c for c in report["cap_curve"] if abs(c["fraction"] - best["optimizer_fraction"]) < 1e-9) md = [] md.append("# VIOLET base-fraction sizing study") md.append("") md.append(f"Generated: `{report['generated_at']}`") md.append("") md.append("## Bottom line") md.append( f"- Recommended fraction: `{best['recommended_fraction']:.3f}` " f"(best Calmar among zero-clip fractions)" ) md.append( f"- Base 0.20 final capital: `{base['final_capital']:.2f}`; " f"recommended final capital: `{best['recommended_final_capital']:.2f}`" ) md.append( f"- Base 0.20 maxDD: `{_format_pct(base['max_drawdown'])}`; " f"recommended maxDD: `{_format_pct(best['recommended_max_drawdown'])}`" ) md.append( f"- Recommended clip rate: `{recommended_cap['pct_clipped']:.2f}%`; " f"unconstrained optimizer: `{best['optimizer_fraction']:.3f}` with " f"`{optimizer_cap['pct_clipped']:.2f}%` clipped at the 3x ceiling." ) md.append("") md.append("## Caveat") md.append( "- Direct non-null slippage telemetry was not available in `trade_execution_quality`; " "the study uses a conservative taker-heavy impact proxy." ) md.append("") md.append("## Cap binding") md.append( f"- The recommended fraction hits the 3x translator cap on `{recommended_cap['pct_clipped']:.2f}%` of trades." ) md.append("") md.append("## Kelly anchor") md.append( f"- Empirical Kelly anchor: `{report['kelly']['kelly_fraction']:.3f}`; " f"fractional anchor: `{report['kelly']['fractional_kelly_anchor']:.3f}`" ) md.append("") md.append("## Files") md.append(f"- JSON: `{json_path}`") md.append(f"- Markdown: `{md_path}`") md_path.write_text("\n".join(md) + "\n") return json_path, md_path def self_test() -> None: trades = [ TradeRow( trade_id="t1", ts=datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc), asset="X", side="SHORT", entry_price=100.0, exit_price=99.0, quantity=1.0, pnl=10.0, pnl_pct=0.10, exit_reason="FIXED_TP", leverage=2.0, capital_before=100.0, capital_after=110.0, bars_held=5, regime_signal=-1, vel_div_entry=-0.03, boost_at_entry=1.0, beta_at_entry=1.0, posture="APEX", our_leverage=0.4, composite_hash=1, notional_quote=100.0, taker_fill_rows=1, fill_rows=1, ), TradeRow( trade_id="t2", ts=datetime(2026, 1, 2, 0, 0, tzinfo=timezone.utc), asset="X", side="SHORT", entry_price=100.0, exit_price=101.0, quantity=1.0, pnl=-5.0, pnl_pct=-0.01, exit_reason="MAX_HOLD", leverage=7.0, capital_before=110.0, capital_after=105.0, bars_held=5, regime_signal=-1, vel_div_entry=-0.03, boost_at_entry=1.0, beta_at_entry=1.0, posture="APEX", our_leverage=0.6, composite_hash=1, notional_quote=110.0, taker_fill_rows=1, fill_rows=1, ), ] rep = replay_equity(trades, 0.20, slippage_enabled=False, median_notional=100.0, ruin_threshold=0.50) assert rep["n_trades"] == 2 assert rep["pct_trades_clipped_at_3x"] == 0.0 assert rep["final_capital"] > 100.0 cap = cap_binding_curve(trades, [0.20, 0.50]) assert cap[0]["pct_clipped"] == 0.0 assert cap[1]["pct_clipped"] == 50.0 slip = slippage_bps_model(100.0, 100.0) assert slip > 0.0 def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser() parser.add_argument("--self-test", action="store_true", help="run the deterministic synthetic fixture and exit") parser.add_argument("--dry-run", action="store_true", help="run the live analysis but do not write files") args = parser.parse_args(argv) if args.self_test: self_test() print("self-test ok") return 0 report = analyze() if args.dry_run: print(json.dumps(report["recommendation"], indent=2, sort_keys=True, default=str)) return 0 json_path, md_path = write_report(report) print(json_path) print(md_path) print(json.dumps(report["recommendation"], indent=2, sort_keys=True, default=str)) return 0 if __name__ == "__main__": raise SystemExit(main())