Cleanup retired S/R research scaffolding

- Remove unused _gate_eligible_levels filtering logic and its tests (research-only)
- Add prominent RESEARCH/DIAGNOSTIC markers and docs to clear-air/ATR fallback helpers
- Document production vs research BACKTEST_* environment variables in backtest_service
- Minor cleanups: update legacy report text, improve outdated function docstring
This commit is contained in:
2026-07-13 18:52:24 +02:00
parent 9f06304100
commit bddaeb9110
4 changed files with 33 additions and 79 deletions
+32 -22
View File
@@ -18,6 +18,18 @@ after D to record the realized outcome. The report contains:
Limitation: sentiment and fundamentals have no point-in-time history, so they're Limitation: sentiment and fundamentals have no point-in-time history, so they're
held neutral here — this calibrates the price/S-R machinery only. held neutral here — this calibrates the price/S-R machinery only.
Environment variables (see also run_backtest_snapshot.py):
Production / general use:
BACKTEST_HOLDOUT_SPLIT=YYYY-MM-DD # disjoint train/test split
BACKTEST_SNAPSHOT_OFFLINE=1
BACKTEST_ALLOW_SPAWN=1 # for Windows multiprocessing
Research / diagnostic only (retired experiments — do not use for live decisions):
BACKTEST_ATR_TARGET_FALLBACK=3
BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1
BACKTEST_RESEARCH_EXITS=1
BACKTEST_MIN_RR_SWEEP=1
""" """
from __future__ import annotations from __future__ import annotations
@@ -67,7 +79,6 @@ from app.services.qualification import (
from app.services.recommendation_service import ( from app.services.recommendation_service import (
_choose_recommended_action, _choose_recommended_action,
_classify_by_probability, _classify_by_probability,
_gate_eligible_levels,
_prune_floor_pinned_targets, _prune_floor_pinned_targets,
_risk_level_from_conflicts, _risk_level_from_conflicts,
_select_primary_target, _select_primary_target,
@@ -145,13 +156,20 @@ def validate_backtest_target_model(value: str) -> str:
return normalized return normalized
# ---------------------------------------------------------------------------
# RESEARCH / DIAGNOSTIC FALLBACKS (retired experiments)
#
# These implement behavior from experiments that were rejected for production
# (clear-air synthetic targets, blanket ATR fallbacks). They are OFF by default
# and exist only to reproduce historical research results or run future ablations.
# See docs/research/sr-levels-and-exits.md.
# Do NOT enable for production decision making.
# ---------------------------------------------------------------------------
def _atr_target_fallback_k() -> float | None: def _atr_target_fallback_k() -> float | None:
"""Research ablation: k for a synthetic k*ATR target when a direction has no """RESEARCH DIAGNOSTIC: k for a synthetic k*ATR target when no S/R level.
S/R level to aim at. Off (None) by default, which is production behavior Off (None) by default (production behavior). Set BACKTEST_ATR_TARGET_FALLBACK=3
no resistance above means no long setup at all. That veto lands hardest on to enable. See docs/research/sr-levels-and-exits.md."""
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() raw = os.getenv("BACKTEST_ATR_TARGET_FALLBACK", "").strip()
if not raw: if not raw:
return None return None
@@ -163,13 +181,8 @@ def _atr_target_fallback_k() -> float | None:
def _fallback_clear_air_only() -> bool: def _fallback_clear_air_only() -> bool:
"""Restrict the fallback to setups with genuinely NO structure ahead. """RESEARCH DIAGNOSTIC: restrict fallback to genuine clear-air cases only.
Set BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1."""
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 { return os.getenv("BACKTEST_FALLBACK_CLEAR_AIR_ONLY", "").strip().lower() in {
"1", "true", "yes", "on", "1", "true", "yes", "on",
} }
@@ -190,11 +203,7 @@ def _has_structure_ahead(direction: str, entry: float, sr_levels: list[Any]) ->
def _atr_fallback_target( def _atr_fallback_target(
direction: str, entry: float, stop: float, atr: float, k: float direction: str, entry: float, stop: float, atr: float, k: float
) -> dict: ) -> dict:
"""A synthetic target k*ATR from entry, shaped like a TargetGenerator row. """RESEARCH DIAGNOSTIC: synthetic target k*ATR (neutral strength)."""
``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 price = entry + k * atr if direction == "long" else entry - k * atr
distance = abs(price - entry) distance = abs(price - entry)
risk = abs(entry - stop) risk = abs(entry - stop)
@@ -254,7 +263,7 @@ def _window_setups(
if not sr_levels: if not sr_levels:
return [] return []
gate_levels = _gate_eligible_levels(sr_levels) gate_levels = list(sr_levels)
technical = (compute_technical_from_arrays(highs, lows, closes, volumes)[0]) or 50.0 technical = (compute_technical_from_arrays(highs, lows, closes, volumes)[0]) or 50.0
momentum = (compute_momentum_from_closes(closes)[0]) or 50.0 momentum = (compute_momentum_from_closes(closes)[0]) or 50.0
@@ -280,8 +289,9 @@ def _window_setups(
fallback_k = _atr_target_fallback_k() fallback_k = _atr_target_fallback_k()
if fallback_k is None: if fallback_k is None:
continue continue
# RESEARCH DIAGNOSTIC only (see _atr_target_fallback_k etc.)
if _fallback_clear_air_only() and _has_structure_ahead(direction, entry, sr_levels): 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 continue
targets = [_atr_fallback_target(direction, entry, stop, atr, fallback_k)] targets = [_atr_fallback_target(direction, entry, stop, atr, fallback_k)]
for t in targets: for t in targets:
t["probability"] = probability_estimator.estimate_probability( t["probability"] = probability_estimator.estimate_probability(
@@ -2452,7 +2462,7 @@ def _sharpe_key(row: dict) -> float:
def _build_research_recommendation(report: dict) -> dict: def _build_research_recommendation(report: dict) -> dict:
"""Advisory rules for the remaining research variants after residual promotion.""" """Build advisory notes from any strategy variants present in the report."""
variants = { variants = {
v.get("variant"): v v.get("variant"): v
for v in (report.get("strategy_variants") or {}).get("variants", []) for v in (report.get("strategy_variants") or {}).get("variants", [])
-32
View File
@@ -56,38 +56,6 @@ def _clamp(value: float, low: float, high: float) -> float:
return max(low, min(high, value)) return max(low, min(high, value))
def _gate_eligible_levels(
sr_levels: list[Any],
*,
confirmed_rounds_only: bool = False,
exclude_standalone_rounds: bool = False,
min_round_rejections: int = 2,
) -> list[Any]:
"""Return structures allowed to influence entry qualification.
Round numbers remain useful visual landmarks, but an untouched standalone
round number is not observed market structure. Research variants can require
either confluence with a pivot/volume source or distinct rejection clusters
before such a level is allowed to manufacture a gate target.
"""
if not confirmed_rounds_only and not exclude_standalone_rounds:
return list(sr_levels)
eligible: list[Any] = []
for level in sr_levels:
sources = set(getattr(level, "sources", None) or [
getattr(level, "detection_method", "unknown")
])
is_round_only = sources == {"round_number"}
rejections = int(getattr(level, "rejection_count", 0) or 0)
if exclude_standalone_rounds and is_round_only:
continue
if confirmed_rounds_only and is_round_only and rejections < min_round_rejections:
continue
eligible.append(level)
return eligible
def _zone_representative_levels( def _zone_representative_levels(
sr_levels: list[SRLevel], sr_levels: list[SRLevel],
entry_price: float, entry_price: float,
@@ -263,7 +263,7 @@ export function BacktestPanel() {
)} )}
{' '}· target model:{' '} {' '}· target model:{' '}
<span className={report.params.is_production_target_model === false ? 'text-amber-300' : 'text-blue-300'}> <span className={report.params.is_production_target_model === false ? 'text-amber-300' : 'text-blue-300'}>
{report.params.target_model_label ?? 'Legacy report (model not recorded)'} {report.params.target_model_label ?? 'Unknown (legacy report)'}
</span> </span>
</p> </p>
-24
View File
@@ -5,7 +5,6 @@ from dataclasses import dataclass
from app.services.recommendation_service import ( from app.services.recommendation_service import (
_build_reasoning, _build_reasoning,
_choose_recommended_action, _choose_recommended_action,
_gate_eligible_levels,
_prune_floor_pinned_targets, _prune_floor_pinned_targets,
_select_primary_target, _select_primary_target,
direction_analyzer, direction_analyzer,
@@ -354,26 +353,3 @@ def test_zone_representative_levels_soft_strength_avoids_resaturation():
assert reps[0].strength == 65 assert reps[0].strength == 65
assert set(reps[0].sources) == {"pivot_point", "round_number"} assert set(reps[0].sources) == {"pivot_point", "round_number"}
assert reps[0].rejection_count == 3 assert reps[0].rejection_count == 3
def test_gate_requires_confirmation_for_standalone_round_number():
from types import SimpleNamespace
untouched = SimpleNamespace(
detection_method="round_number", sources=["round_number"],
rejection_count=1,
)
confirmed = SimpleNamespace(
detection_method="round_number", sources=["round_number"],
rejection_count=2,
)
confluent = SimpleNamespace(
detection_method="merged", sources=["round_number", "pivot_point"],
rejection_count=0,
)
assert _gate_eligible_levels(
[untouched, confirmed, confluent], confirmed_rounds_only=True
) == [confirmed, confluent]
assert _gate_eligible_levels(
[untouched, confirmed, confluent], exclude_standalone_rounds=True
) == [confluent]