dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Research: S/R levels, the target exit, and the entry gate

Browse Source 
Investigated whether our support/resistance detection follows best practice
and whether we actually use it that way. Three findings, all backed by runs
against the prod snapshot and written up in docs/research/sr-levels-and-exits.md:

- The S/R target must NOT become an exit. Honoring it as a take-profit on top
  of the 3x ATR trail drops Sharpe 2.04 -> 1.47 and halves CAGR. Win rate rises
  (37.5% -> 40.0%), which is the tell: it truncates the right tail where
  momentum's edge lives.
- The clear-air fallback (synthesize a 3xATR target where no resistance exists,
  so 52-week-high breakouts stop being vetoed) looked strictly better in-sample
  (Sharpe 2.04 -> 2.07, CAGR 50.4% -> 62.3%, DD 21.4% -> 20.1%) but FAILED a
  real out-of-sample holdout: on entries after 2024-07-01 it is worse on Sharpe
  (2.78 -> 2.45) and Calmar, better only on raw CAGR. Not shipped.
- The detector itself is weak vs best practice (POC/VAH/VAL computed then
  discarded, HVN = any above-mean bin, 1.48x volume double-counting, "touch"
  counts pass-throughs, no round numbers), but its only causal path to P&L is
  the entry gate. Fix it for the displayed levels, not for returns.

Method note: nested lookback windows are NOT out-of-sample. The in-sample result
was clean, large, and consistent across five windows, and still did not survive
a proper entry-date split.

All research paths are off by default and the default report is unchanged:
  BACKTEST_RESEARCH_EXITS=1        take-profit exit rows
  BACKTEST_ATR_TARGET_FALLBACK=k   synthetic k*ATR target when S/R offers none
  BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1  restrict that to genuinely clear air
  BACKTEST_HOLDOUT_SPLIT=YYYY-MM-DD   train/test split by entry date

Also fixes two reproducibility holes found while reconciling our local baseline
against the live report:

- create_backtest_snapshot.py now copies paper_% settings. The production
  monitor row replays the runtime exit policy via get_exit_policy(); without
  those keys a snapshot silently falls back to code defaults, so a live-tuned
  exit would never be reflected.
- Migration 020 drops activation_min_expected_value and
  activation_min_target_probability. Both are orphans of the June EV-gate
  redesign, read by no code path, but prod carries min_target_probability = 50.0
  which implies a probability floor that is not enforced (the real floor is the
  20% constant in qualification.py).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
dennisthiessen
2026-07-12 15:05:34 +02:00
co-authored by Claude Opus 4.8
parent fa26ec3ec4
commit 85b3ef618f
9 changed files with 554093 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/backtest_service.py
+206 -7
View File
@@ -123,6 +123,70 @@ def _wrap_levels(level_dicts: list[dict]) -> list[Any]:
]
def _atr_target_fallback_k() -> float | None:
"""Research ablation: k for a synthetic k*ATR target when a direction has no
S/R level to aim at. Off (None) by default, which is production behavior —
no resistance above means no long setup at all. That veto lands hardest on
names at 52-week highs (clear air above), i.e. exactly what the momentum gate
selects, so this flag exists to measure what the veto costs. Set
BACKTEST_ATR_TARGET_FALLBACK=3 to enable. See docs/research/sr-levels-and-exits.md."""
raw = os.getenv("BACKTEST_ATR_TARGET_FALLBACK", "").strip()
if not raw:
return None
try:
k = float(raw)
except ValueError:
return None
return k if k > 0 else None
def _fallback_clear_air_only() -> bool:
"""Restrict the fallback to setups with genuinely NO structure ahead.
Without this, the fallback also fires when levels DO exist ahead but
``TargetGenerator``'s distance filters rejected them (nearer than 1 ATR, or
past ``max_atr_multiple``). Measured on the snapshot, that's 65% of what the
fallback admits — a different population from the clear-air breakouts, which
confounds the famine test. Set BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1."""
return os.getenv("BACKTEST_FALLBACK_CLEAR_AIR_ONLY", "").strip().lower() in {
"1", "true", "yes", "on",
}
def _has_structure_ahead(direction: str, entry: float, sr_levels: list[Any]) -> bool:
"""Is there any S/R level in the direction of the trade? (Resistance above for
a long, support below for a short.) False == clear air."""
if direction == "long":
return any(
lv.type == "resistance" and float(lv.price_level) > entry for lv in sr_levels
)
return any(
lv.type == "support" and float(lv.price_level) < entry for lv in sr_levels
)
def _atr_fallback_target(
direction: str, entry: float, stop: float, atr: float, k: float
) -> dict:
"""A synthetic target k*ATR from entry, shaped like a TargetGenerator row.
``sr_strength`` is 50 (neutral) so the probability model's strength magnet
contributes nothing — the target stands on distance alone.
"""
price = entry + k * atr if direction == "long" else entry - k * atr
distance = abs(price - entry)
risk = abs(entry - stop)
return {
"price": float(price),
"distance_from_entry": float(distance),
"distance_atr_multiple": float(k),
"rr_ratio": float(distance / risk) if risk > 0 else 0.0,
"classification": "Moderate",
"sr_level_id": -1, # synthetic: no S/R level behind it
"sr_strength": 50.0,
}
def _window_setups(
window_records: list,
config: dict,
@@ -174,7 +238,12 @@ def _window_setups(
zone_levels = _zone_representative_levels(sr_levels, entry)
targets = target_generator.generate_targets(direction, entry, stop, zone_levels, atr)
if not targets:
continue
fallback_k = _atr_target_fallback_k()
if fallback_k is None:
continue
if _fallback_clear_air_only() and _has_structure_ahead(direction, entry, sr_levels):
continue # structure exists ahead; the distance filters rejected it, not the famine
targets = [_atr_fallback_target(direction, entry, stop, atr, fallback_k)]
for t in targets:
t["probability"] = probability_estimator.estimate_probability(
t, dim_scores, None, direction, config
@@ -1099,6 +1168,7 @@ def _simulate_portfolio(
risk_per_trade: float = SIM_RISK_PER_TRADE,
atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER,
start_date: date | None = None,
end_date: date | None = None,
include_curve: bool = False,
) -> dict | None:
"""Replay the qualified setups as ONE capital-constrained book and report
@@ -1109,9 +1179,11 @@ def _simulate_portfolio(
``exit_policy``: "target" races the S/R target against the stop with a
timeout at ``hold_days``; "hold" keeps only the initial stop and exits at
the ``hold_days``-th close. Research exits add price-derived early exits:
"sma50", "low20", "technical40", and "atr_trail3". Stops fill at the
worse of stop or open (gaps modeled); positions still open at the end are
closed at their last mark. Returns None when there is nothing to trade.
"sma50", "low20", "technical40", and "atr_trail3". "atr_trail3_target"
runs the ATR trail *and* the S/R take-profit together — the trade ends at
whichever comes first. Stops fill at the worse of stop or open (gaps
modeled); positions still open at the end are closed at their last mark.
Returns None when there is nothing to trade.
"""
if qualified_fn is None:
def _default_qualified(c: dict) -> bool:
@@ -1121,12 +1193,15 @@ def _simulate_portfolio(
entries_by_ord: dict[int, list[dict]] = defaultdict(list)
start_ord = start_date.toordinal() if start_date is not None else None
end_ord = end_date.toordinal() if end_date is not None else None
for c in candidates:
if not qualified_fn(c) or c.get("direction") != "long":
continue
entry_ord = date.fromisoformat(c["date"]).toordinal()
if start_ord is not None and entry_ord < start_ord:
continue
if end_ord is not None and entry_ord >= end_ord:
continue # holdout: entries strictly before the split
if not c.get("entry") or not c.get("stop"):
continue
entries_by_ord[entry_ord].append(c)
@@ -1248,7 +1323,7 @@ def _simulate_portfolio(
)
_close_trade(sym, min(pos["stop"], bar.open), reason)
continue
if exit_policy == "target" and pos["target"] and bar.high >= pos["target"]:
if exit_policy in ("target", "atr_trail3_target") and pos["target"] and bar.high >= pos["target"]:
_close_trade(sym, pos["target"], "target")
continue
if exit_policy == "sma50":
@@ -1269,7 +1344,7 @@ def _simulate_portfolio(
if pos["bars_held"] >= hold_days:
_close_trade(sym, bar.close, "time")
continue
if exit_policy == "atr_trail3":
if exit_policy in ("atr_trail3", "atr_trail3_target"):
pos["highest_close"] = max(pos["highest_close"], bar.close)
atr = _atr(sym, bar.idx)
if atr is not None:
@@ -1752,6 +1827,32 @@ EXIT_POLICY_VARIANTS: tuple[dict, ...] = (
},
)
# Take-profit exits, tested 2026-07-12 and REJECTED — see
# docs/research/sr-levels-and-exits.md. Honoring the S/R target is the worst
# exit of the seven: it lifts the win rate but truncates the right tail where
# momentum's edge lives. Kept so the result stays reproducible, off by default.
# Set BACKTEST_RESEARCH_EXITS=1 to include them in the exit comparison.
RESEARCH_EXIT_POLICY_VARIANTS: tuple[dict, ...] = (
{
"exit_policy": "target",
"label": "80/20 entry + S/R target take-profit (no trail)",
"description": "Take profit at the S/R target; initial stop only, no trailing.",
},
{
"exit_policy": "atr_trail3_target",
"label": "80/20 entry + 3x ATR trail AND S/R target take-profit",
"description": "Production trail plus a take-profit at the S/R target, first one wins.",
},
)
def _exit_policy_variants() -> tuple[dict, ...]:
"""The exit book. Research take-profit rows only when explicitly opted in, so
the default report stays identical to the shipped baseline."""
if os.getenv("BACKTEST_RESEARCH_EXITS", "").strip().lower() in {"1", "true", "yes", "on"}:
return EXIT_POLICY_VARIANTS + RESEARCH_EXIT_POLICY_VARIANTS
return EXIT_POLICY_VARIANTS
PORTFOLIO_MONITOR_LOOKBACKS: tuple[dict, ...] = (
{"lookback": "6m", "label": "6 months", "days": 183},
@@ -1812,7 +1913,7 @@ def _exit_policy_sims(
rows: list[dict] = []
ranking_key = str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"])
for cfg in EXIT_POLICY_VARIANTS:
for cfg in _exit_policy_variants():
sim = _simulate_portfolio(
candidates,
prices,
@@ -1843,6 +1944,96 @@ def _lookback_start(max_ord: int | None, days: int | None) -> date | None:
return date.fromordinal(max_ord - days)
def _holdout_split() -> date | None:
"""Train/test split date for out-of-sample validation, e.g.
BACKTEST_HOLDOUT_SPLIT=2024-07-01. Off by default."""
raw = os.getenv("BACKTEST_HOLDOUT_SPLIT", "").strip()
if not raw:
return None
try:
return date.fromisoformat(raw)
except ValueError:
return None
def _holdout_evaluation(
candidates: list[dict],
prices: dict[str, tuple],
_spy_closes: dict[date, float] | None,
hold_days: int,
split: date,
live_exit_policy: dict | None = None,
) -> dict:
"""The production strategy simulated on entries BEFORE the split (train) and
on entries ON/AFTER it (test), as separate books.
The lookback rows in ``portfolio_monitor`` are NOT a holdout — they are nested
windows that all end today, so every one of them overlaps the data a rule was
chosen on. This does the real thing: the test window is disjoint from the
train window, so a rule settled on train has never seen it.
"""
strategy = next(
(s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")),
None,
)
if strategy is None:
return {}
entry_cfg = _entry_variant_config(str(strategy["entry_variant"]))
if entry_cfg is None:
return {}
exit_policy = str(strategy["exit_policy"])
row_hold_days = hold_days
trail_multiplier = ATR_TRAIL_MULTIPLIER
if strategy.get("use_live_config") and live_exit_policy is not None:
exit_policy = LIVE_EXIT_MODE_TO_SIM.get(
str(live_exit_policy.get("mode", "atr_trailing")), "atr_trail3"
)
row_hold_days = int(live_exit_policy.get("hold_days", hold_days))
trail_multiplier = float(
live_exit_policy.get("atr_multiplier", ATR_TRAIL_MULTIPLIER)
)
qualified_fn = (
None if strategy.get("use_live_config")
else lambda c, config=entry_cfg: _qualifies_strategy_variant(c, config)
)
rows: list[dict] = []
for window, start, end in (
("train", None, split),
("test", split, None),
):
sim = _simulate_portfolio(
candidates,
prices,
_spy_closes,
exit_policy,
row_hold_days,
qualified_fn=qualified_fn,
ranking_key=str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"]),
max_positions=int(entry_cfg["max_positions"]),
risk_per_trade=float(entry_cfg["risk_per_trade"]),
atr_trail_multiplier=trail_multiplier,
start_date=start,
end_date=end,
include_curve=True,
)
if sim is None:
continue
rows.append({"window": window, "exit_policy": exit_policy, **sim})
return {
"split_date": split.isoformat(),
"strategy": strategy["strategy"],
"rows": rows,
"note": (
"Train = entries before the split; test = entries on/after it. The two "
"books are disjoint in entry date. Compare the TEST row across arms: a "
"rule chosen by looking at full history has already seen train."
),
}
def _portfolio_monitor(
candidates: list[dict],
prices: dict[str, tuple],
@@ -2433,6 +2624,7 @@ async def run_backtest(
strategy_variant_rows: list[dict] = []
exit_policy_rows: list[dict] = []
portfolio_monitor_report: dict | None = None
holdout_report: dict | None = None
try:
qual_symbols = sorted({
c["symbol"]
@@ -2481,6 +2673,12 @@ async def run_backtest(
candidates, price_columns, spy_closes, hold_horizon,
live_exit_policy=live_exit_policy,
)
split = _holdout_split()
if split is not None:
holdout_report = _holdout_evaluation(
candidates, price_columns, spy_closes, hold_horizon, split,
live_exit_policy=live_exit_policy,
)
except Exception:
logger.exception("Portfolio simulation failed")
@@ -2555,6 +2753,7 @@ async def run_backtest(
),
},
"portfolio_monitor": portfolio_monitor_report,
"holdout": holdout_report,
"signal_eval": _signal_evaluation(collected),
"signal_eval_note": (
"Cross-sectional rank-IC of price-only signals vs the forward "