fix: symbol-specific Kelly win rate, wire dead sizing code, safer walk-forward fallback
- signal_booster.py: _compute_rates() now also computes a per-symbol win rate (not just system-wide/direction aggregates); trade_executor.py's Kelly sizing prefers it when the symbol has enough closed-trade history. Added _as_datetime() to normalize closed_at across backends/drivers that return either a real datetime or a string from raw SQL. - trade_executor.py: volatility-filter and Kelly-sizing exception handlers now log at warning level with the actual exception instead of silently swallowing failures that affect how much money a trade risks. - risk_manager.py: compute_partial_tp_levels() now returns all 3 levels its docstring always promised (TP1 25% + TP2 35% + 40% trailing remainder) instead of silently dropping the last 40%. - trade_executor.py: compute_volatility_adjusted_size() was dead code; now applied as a multiplier on the Kelly-derived trade_size (using max_risk_pct=100 to reinterpret it as "scale the already-sized trade" rather than "% of a bankroll", which would always collapse to this pipeline's $5 floor at its actual dollar scale). - walk_forward.py: grid-search fallback (when every combo is too sparse to trust) now picks the combo with the most trades/highest PnL instead of always the grid's arbitrary first entry. Raised MIN_TRADES_PER_FOLD 5 -> 15 for a more defensible statistical minimum. 219 backend tests pass (+10). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Le
@@ -179,6 +179,12 @@ class AdaptiveSLTPOptimizer:
|
||||
- TP1: ATR × 2.0 → close 25%
|
||||
- TP2: ATR × 4.0 → close 35%
|
||||
- Remainder: 40% with trailing stop
|
||||
|
||||
The three `close_percentage`s always sum to 1.0 (100% of the
|
||||
position) — the first two are fixed-price take-profit levels, the
|
||||
third has `price: None` and `trailing: True` since "trail the
|
||||
remainder" is a runtime stop-management decision, not a price
|
||||
this function can compute in isolation.
|
||||
"""
|
||||
params = REGIME_MULTIPLIERS.get(regime, REGIME_MULTIPLIERS["neutral"])
|
||||
direction = direction.upper()
|
||||
@@ -198,6 +204,14 @@ class AdaptiveSLTPOptimizer:
|
||||
result.append({
|
||||
"price": round(price, 8),
|
||||
"close_percentage": level["close_pct"],
|
||||
"trailing": False,
|
||||
})
|
||||
|
||||
remainder_pct = 1.0 - sum(level["close_pct"] for level in levels)
|
||||
result.append({
|
||||
"price": None,
|
||||
"close_percentage": round(remainder_pct, 8),
|
||||
"trailing": True,
|
||||
})
|
||||
|
||||
return result
|
||||
|
||||
@@ -50,6 +50,31 @@ _STRATEGY_MAP: dict[str, str] = {
|
||||
}
|
||||
|
||||
|
||||
def _as_datetime(value: Any) -> datetime | None:
|
||||
"""Normalize a raw `closed_at` value from `text()` SQL to a datetime.
|
||||
|
||||
Raw textual SQL (unlike ORM queries) carries no column type
|
||||
information, so the driver returns whatever native type it stores
|
||||
timestamps as — Postgres/asyncpg gives back a real `datetime`, but
|
||||
SQLite/aiosqlite (used in this project's test suite) gives back a
|
||||
plain string. Without this, the exponential-decay weighting below
|
||||
would raise on any backend/driver that doesn't hand back a `datetime`.
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
value = datetime.fromisoformat(value)
|
||||
except ValueError:
|
||||
return None
|
||||
if not isinstance(value, datetime):
|
||||
return None
|
||||
# This app stores/consumes timestamps as UTC throughout — a
|
||||
# driver/backend that hands back a naive datetime (e.g. SQLite) means
|
||||
# "UTC with the tzinfo stripped", not "some unspecified local time".
|
||||
return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
# ── Core helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
async def compute_strategy_win_rates(db: AsyncSession | None = None) -> dict[str, float]:
|
||||
@@ -109,8 +134,8 @@ async def _compute_rates(db: AsyncSession) -> dict[str, float]:
|
||||
for row in rows:
|
||||
strategy = str(row[0])
|
||||
is_win = int(row[1])
|
||||
closed_at = row[2]
|
||||
|
||||
closed_at = _as_datetime(row[2])
|
||||
|
||||
if closed_at:
|
||||
days_ago = (now_dt - closed_at).days
|
||||
weight = math.exp(-DECAY_LAMBDA * max(days_ago, 0))
|
||||
@@ -120,10 +145,13 @@ async def _compute_rates(db: AsyncSession) -> dict[str, float]:
|
||||
strategy_weights[strategy] = strategy_weights.get(strategy, 0) + weight
|
||||
strategy_wins[strategy] = strategy_wins.get(strategy, 0) + (weight if is_win else 0)
|
||||
|
||||
# Also compute direction-specific rates from the same data
|
||||
# Also compute direction- and symbol-specific rates from the same
|
||||
# data in one query (fix mm: Kelly sizing prefers a symbol's own
|
||||
# win rate over the system-wide aggregate when there's enough
|
||||
# history for that specific symbol).
|
||||
dir_result = await db.execute(
|
||||
text("""
|
||||
SELECT direction,
|
||||
SELECT direction, symbol,
|
||||
CASE WHEN pnl > 0 THEN 1 ELSE 0 END,
|
||||
closed_at
|
||||
FROM hypothetical_trades
|
||||
@@ -133,10 +161,13 @@ async def _compute_rates(db: AsyncSession) -> dict[str, float]:
|
||||
LIMIT 5000
|
||||
"""),
|
||||
)
|
||||
symbol_weights: dict[str, float] = {}
|
||||
symbol_wins: dict[str, float] = {}
|
||||
for row in dir_result.all():
|
||||
direction = str(row[0]) if row[0] else "UNKNOWN"
|
||||
is_win = int(row[1])
|
||||
closed_at = row[2]
|
||||
symbol = str(row[1]) if row[1] else None
|
||||
is_win = int(row[2])
|
||||
closed_at = _as_datetime(row[3])
|
||||
if closed_at:
|
||||
days_ago = (now_dt - closed_at).days
|
||||
weight = math.exp(-DECAY_LAMBDA * max(days_ago, 0))
|
||||
@@ -144,6 +175,9 @@ async def _compute_rates(db: AsyncSession) -> dict[str, float]:
|
||||
weight = 0.5
|
||||
direction_weights[direction] = direction_weights.get(direction, 0) + weight
|
||||
direction_wins[direction] = direction_wins.get(direction, 0) + (weight if is_win else 0)
|
||||
if symbol:
|
||||
symbol_weights[symbol] = symbol_weights.get(symbol, 0) + weight
|
||||
symbol_wins[symbol] = symbol_wins.get(symbol, 0) + (weight if is_win else 0)
|
||||
|
||||
rates: dict[str, float] = {}
|
||||
total_weight = 0.0
|
||||
@@ -172,6 +206,15 @@ async def _compute_rates(db: AsyncSession) -> dict[str, float]:
|
||||
if dw >= MIN_TRADES * 0.5:
|
||||
rates[f"__all___{direction}"] = ww / dw
|
||||
|
||||
# Symbol-specific rates (fix mm) — same MIN_TRADES bar as everything
|
||||
# else here, so a thinly-traded symbol falls back to the direction/
|
||||
# overall aggregate instead of a noisy few-trade estimate.
|
||||
for symbol in symbol_weights:
|
||||
sw = symbol_weights.get(symbol, 0)
|
||||
sww = symbol_wins.get(symbol, 0)
|
||||
if sw >= MIN_TRADES * 0.5:
|
||||
rates[f"__symbol__{symbol}"] = sww / sw
|
||||
|
||||
_win_rate_cache = rates
|
||||
_last_cache_update = datetime.now(timezone.utc)
|
||||
await redis_client.set_json(_REDIS_KEY_WIN_RATES, rates, _CACHE_TTL_SECONDS)
|
||||
|
||||
@@ -171,19 +171,29 @@ async def execute_signal_trade(
|
||||
continue
|
||||
|
||||
# ── Volatility filter ──
|
||||
# atr_pct_for_sizing survives past this block (fix oo) so the Kelly
|
||||
# sizing step below can scale size down gracefully between the two
|
||||
# skip thresholds, instead of ATR only ever being a binary
|
||||
# skip/allow gate with no effect in between.
|
||||
atr_pct_for_sizing: float | None = None
|
||||
try:
|
||||
snap = json.loads(signal.indicators_snapshot) if signal.indicators_snapshot else {}
|
||||
atr_val = snap.get("atr_14")
|
||||
if atr_val and isinstance(atr_val, list) and len(atr_val) > 0 and atr_val[-1]:
|
||||
atr_pct = float(atr_val[-1]) / float(current_price) * 100
|
||||
atr_pct_for_sizing = atr_pct
|
||||
if atr_pct > 8.0:
|
||||
logger.info("⛔ Skipping %s — ATR too high: %.2f%%", symbol, atr_pct)
|
||||
continue
|
||||
if atr_pct < 0.5:
|
||||
logger.info("⛔ Skipping %s — ATR too low: %.2f%%", symbol, atr_pct)
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
# A parse/format error here must not silently disable the
|
||||
# volatility gate in production with no trace of why — surface
|
||||
# it, even though we still proceed (fail-open, matching the
|
||||
# existing behavior of not blocking a trade over a data glitch).
|
||||
logger.warning("Volatility filter failed for %s, proceeding without it: %s", symbol, e)
|
||||
|
||||
# ── Serialize per-user trade-opening decisions (fix ll) ──
|
||||
# The `with_for_update()` on all_open_trades below only locks rows
|
||||
@@ -288,12 +298,19 @@ async def execute_signal_trade(
|
||||
kelly = DynamicKellySizer()
|
||||
overall_rate = rates.get("__all__", 0.5)
|
||||
dir_rate = rates.get(f"__all___{signal_direction}", overall_rate)
|
||||
# (mm) Prefer this specific symbol's own win rate when there's
|
||||
# enough history for it — a coin that trades very differently
|
||||
# from the system-wide average (e.g. a consistently weaker
|
||||
# altcoin) should be sized off its own edge, not the aggregate.
|
||||
# Falls back to the direction/overall aggregate exactly as
|
||||
# before when the symbol doesn't have enough closed trades yet.
|
||||
win_rate = rates.get(f"__symbol__{symbol}", dir_rate)
|
||||
signal_confidence = 0.5
|
||||
if signal.indicators_snapshot:
|
||||
snap = json.loads(signal.indicators_snapshot)
|
||||
signal_confidence = snap.get("confidence", 0.5)
|
||||
kelly_pct = kelly.compute_kelly_pct(
|
||||
win_rate=dir_rate,
|
||||
win_rate=win_rate,
|
||||
avg_win=pnl_stats.get("avg_win", 3.0),
|
||||
avg_loss=pnl_stats.get("avg_loss", 2.0),
|
||||
confidence=signal_confidence,
|
||||
@@ -313,8 +330,39 @@ async def execute_signal_trade(
|
||||
kelly_pct *= 1.0 / math.sqrt(same_direction_open + 1)
|
||||
if kelly_pct > 0:
|
||||
trade_size = max(trade_size * Decimal(str(kelly_pct)), Decimal("1"))
|
||||
except Exception:
|
||||
logger.debug("Kelly sizing failed, using fixed trade_size")
|
||||
|
||||
# ── Volatility/regime adjustment (fix oo) ──
|
||||
# `compute_volatility_adjusted_size` was previously never
|
||||
# called anywhere — the volatility filter above only ever
|
||||
# skipped a trade outright above/below its two hard cutoffs,
|
||||
# with no graduated effect in between. `max_risk_pct=100` here
|
||||
# means "scale 100% of the Kelly-derived trade_size by
|
||||
# volatility/regime" rather than the function's own docstring
|
||||
# framing ("% of a bankroll to risk") — that framing assumes a
|
||||
# much larger base_size (an account balance) than this
|
||||
# paper-trading pipeline's small fixed trade_size, where a
|
||||
# literal 1-2% risk-per-trade would always collapse to the $5
|
||||
# floor below regardless of volatility. Reusing the same
|
||||
# vol_factor/regime_factor math as a pure multiplier on the
|
||||
# already-sized trade instead keeps it meaningful at this
|
||||
# pipeline's actual dollar scale.
|
||||
if atr_pct_for_sizing is not None:
|
||||
regime = "neutral"
|
||||
if signal.indicators_snapshot:
|
||||
regime = snap.get("market_regime") or "neutral"
|
||||
trade_size = kelly.compute_volatility_adjusted_size(
|
||||
base_size=trade_size,
|
||||
atr_pct=Decimal(str(atr_pct_for_sizing)),
|
||||
max_risk_pct=Decimal("100"),
|
||||
regime=regime,
|
||||
)
|
||||
except Exception as e:
|
||||
# Silently falling back here used to hide real bugs in the
|
||||
# sizing pipeline (wrong rates shape, bad Decimal conversion,
|
||||
# etc.) — this affects how much real/paper money a trade
|
||||
# risks, so a failure here should be visible, not just a
|
||||
# debug-level breadcrumb.
|
||||
logger.warning("Kelly sizing failed for %s, using fixed trade_size: %s", symbol, e)
|
||||
|
||||
# Sane size bounds
|
||||
trade_size = max(trade_size, Decimal("5"))
|
||||
|
||||
@@ -47,7 +47,15 @@ DEFAULT_PARAM_GRID: dict[str, list[float]] = {
|
||||
"max_hold_candles": [24, 48, 96],
|
||||
}
|
||||
|
||||
MIN_TRADES_PER_FOLD = 5 # reject param combos too sparse to trust
|
||||
# (fix rr) Reject param combos too sparse to trust. 5 trades is too thin a
|
||||
# sample to estimate a Sharpe-like ratio's mean/std reliably — a couple of
|
||||
# outlier trades can swing it wildly. 15 is still well short of the ~30
|
||||
# quant practitioners often cite for a stable estimate, but demanding 30
|
||||
# per fold would starve most folds of any "trustworthy" combo at all given
|
||||
# this system's selective (STRONG-only) entry signals — 15 is a middle
|
||||
# ground between statistical caution and having enough folds to walk
|
||||
# forward over at all.
|
||||
MIN_TRADES_PER_FOLD = 15
|
||||
WARMUP_BUFFER_CANDLES = 60 # extra history fetched before each window so indicators aren't cold at window start
|
||||
|
||||
_TF_MINUTES = {"15m": 15, "30m": 30, "1h": 60, "4h": 240, "1d": 1440, "1w": 10080, "1M": 43200}
|
||||
@@ -175,7 +183,20 @@ def _grid_search(
|
||||
keys = list(param_grid.keys())
|
||||
best_params: dict[str, float] | None = None
|
||||
best_score = float("-inf")
|
||||
best_stats: dict = {}
|
||||
best_signals: list[dict] | None = None
|
||||
best_trades: list[dict] | None = None
|
||||
|
||||
# (fix xx) If every combo is too sparse to trust (_fold_score returns
|
||||
# -inf for all of them), we still need to report *something* for the
|
||||
# fold — track the least-bad combo as we go instead of always falling
|
||||
# back to the grid's arbitrary first entry, which could easily be the
|
||||
# worst-performing one. Ranked by (trade count, total PnL): more trades
|
||||
# means closer to being statistically trustworthy in the first place,
|
||||
# and PnL breaks ties between equally-sparse combos.
|
||||
fallback_params: dict[str, float] | None = None
|
||||
fallback_rank: tuple[int, float] = (-1, float("-inf"))
|
||||
fallback_signals: list[dict] | None = None
|
||||
fallback_trades: list[dict] | None = None
|
||||
|
||||
for combo in product(*(param_grid[k] for k in keys)):
|
||||
params = dict(zip(keys, combo))
|
||||
@@ -185,15 +206,18 @@ def _grid_search(
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best_params = params
|
||||
best_stats = _compute_stats(all_signals, trades)
|
||||
best_signals, best_trades = all_signals, trades
|
||||
|
||||
rank = (len(closed_trades), sum(float(t.get("pnl", 0.0)) for t in closed_trades))
|
||||
if rank > fallback_rank:
|
||||
fallback_rank = rank
|
||||
fallback_params = params
|
||||
fallback_signals, fallback_trades = all_signals, trades
|
||||
|
||||
if best_params is None:
|
||||
# Every combo scored -inf (too few trades) — still report the
|
||||
# grid's first combination so the fold has *something* to show.
|
||||
best_params = {k: param_grid[k][0] for k in keys}
|
||||
all_signals, trades = _run_combo(candles, scores_series, trade_size, best_params, fee_pct, slippage_pct)
|
||||
best_stats = _compute_stats(all_signals, trades)
|
||||
best_params, best_signals, best_trades = fallback_params, fallback_signals, fallback_trades
|
||||
|
||||
best_stats = _compute_stats(best_signals, best_trades)
|
||||
return best_params, best_score, best_stats
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user