867 lines
32 KiB
Python
867 lines
32 KiB
Python
#!/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())
|