Compare commits

...
3 Commits
27 changed files with 37717 additions and 62 deletions
+1
View File
@@ -387,6 +387,7 @@ async def trigger_job(
db,
job_name,
target_model=body.target_model if body is not None else None,
cadence=body.cadence if body is not None else None,
)
return APIEnvelope(status="success", data=result)
+2
View File
@@ -36,6 +36,7 @@ async def list_trade_setups(
recommended_action=recommended_action,
live_recommendation=True,
exclude_open_trade_tickers=True,
exclude_reentry_lockdown_tickers=True,
)
data = []
@@ -98,6 +99,7 @@ async def get_ticker_trade_setups(
db,
symbol=symbol,
live_recommendation=True,
include_reentry_lockdown=True,
)
data = []
for row in rows:
+37 -10
View File
@@ -37,8 +37,10 @@ from app.services import fundamental_service, ingestion_service, sentiment_servi
from app.services.alert_service import dispatch_alerts
from app.services.backtest_service import (
BACKTEST_TARGET_MODELS,
DEFAULT_BACKTEST_CADENCE,
PRODUCTION_GTL_TARGET_MODEL,
run_and_store as run_backtest_and_store,
validate_backtest_cadence,
validate_backtest_target_model,
)
from app.services.benchmark_service import refresh_benchmark_prices
@@ -112,6 +114,7 @@ def _idle_runtime() -> dict[str, object]:
_job_runtime: dict[str, dict[str, object]] = {name: _idle_runtime() for name in _JOB_NAMES}
_next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
_next_backtest_cadence = DEFAULT_BACKTEST_CADENCE
# ---------------------------------------------------------------------------
@@ -119,23 +122,44 @@ _next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
# ---------------------------------------------------------------------------
def queue_backtest_target_model(target_model: str | None) -> str:
"""Select the model for the next manual backtest run only.
def queue_backtest_options(
target_model: str | None,
cadence: str | None,
) -> tuple[str, str]:
"""Select model and cadence for the next manual backtest run only.
Scheduled runs and subsequent manual runs return to the production GTL.
Scheduled and subsequent manual runs return to production GTL at the
resource-safe weekly cadence.
"""
global _next_backtest_target_model
selected = validate_backtest_target_model(
global _next_backtest_target_model, _next_backtest_cadence
selected_model = validate_backtest_target_model(
target_model or PRODUCTION_GTL_TARGET_MODEL
)
_next_backtest_target_model = selected
selected_cadence = validate_backtest_cadence(
cadence or DEFAULT_BACKTEST_CADENCE
)
_next_backtest_target_model = selected_model
_next_backtest_cadence = selected_cadence
return selected_model, selected_cadence
def queue_backtest_target_model(target_model: str | None) -> str:
"""Compatibility wrapper for callers selecting only the target model."""
selected, _ = queue_backtest_options(target_model, DEFAULT_BACKTEST_CADENCE)
return selected
def _consume_backtest_options() -> tuple[str, str]:
global _next_backtest_target_model, _next_backtest_cadence
selected = (_next_backtest_target_model, _next_backtest_cadence)
_next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
_next_backtest_cadence = DEFAULT_BACKTEST_CADENCE
return selected
def _consume_backtest_target_model() -> str:
global _next_backtest_target_model
selected = _next_backtest_target_model
_next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
"""Compatibility wrapper consuming all queued one-run options."""
selected, _ = _consume_backtest_options()
return selected
@@ -1028,12 +1052,13 @@ async def compute_regime_monitor() -> None:
async def run_backtest_job() -> None:
"""Replay the price-derived engine over history and cache the report."""
job_name = "backtest"
target_model = _consume_backtest_target_model()
target_model, cadence = _consume_backtest_options()
_log_event(
logging.INFO,
"job_start",
job=job_name,
target_model=target_model,
cadence=cadence,
)
_runtime_start(job_name)
@@ -1051,6 +1076,7 @@ async def run_backtest_job() -> None:
db,
_on_progress,
target_model=target_model,
cadence=cadence,
)
_runtime_finish(
@@ -1058,6 +1084,7 @@ async def run_backtest_job() -> None:
processed=report.get("tickers", 0), total=report.get("tickers", 0),
message=(
f"{BACKTEST_TARGET_MODELS[target_model]}: "
f"{cadence} cadence, "
f"{report.get('candidates', 0)} setups, "
f"{report.get('qualified', 0)} qualified"
),
+1
View File
@@ -46,6 +46,7 @@ class JobToggle(BaseModel):
class JobTriggerRequest(BaseModel):
"""Optional parameters for a one-time manual job run."""
target_model: Literal["production_gtl", "structural_sr"] | None = None
cadence: Literal["weekly", "daily"] | None = None
class RecommendationConfigUpdate(BaseModel):
+1
View File
@@ -59,5 +59,6 @@ class TradeSetupResponse(BaseModel):
momentum_percentile: float | None = None
strategy_rank: float | None = None
volatility_percentile: float | None = None
reentry_lockdown_remaining_sessions: int | None = None
context_as_of: TradeSetupContextAsOfResponse | None = None
recommendation_summary: RecommendationSummaryResponse | None = None
+7 -2
View File
@@ -607,6 +607,7 @@ async def trigger_job(
job_name: str,
*,
target_model: str | None = None,
cadence: str | None = None,
) -> dict[str, str]:
"""Trigger a manual job run via the scheduler.
@@ -616,6 +617,8 @@ async def trigger_job(
raise ValidationError(f"Unknown job: {job_name}. Valid jobs: {', '.join(sorted(VALID_JOB_NAMES))}")
if target_model is not None and job_name != "backtest":
raise ValidationError("target_model is supported only for the backtest job")
if cadence is not None and job_name != "backtest":
raise ValidationError("cadence is supported only for the backtest job")
from app.scheduler import get_job_runtime_snapshot, scheduler
@@ -643,9 +646,9 @@ async def trigger_job(
return {"job": job_name, "status": "not_found", "message": f"Job '{job_name}' is not registered in the scheduler"}
if job_name == "backtest":
from app.scheduler import queue_backtest_target_model
from app.scheduler import queue_backtest_options
target_model = queue_backtest_target_model(target_model)
target_model, cadence = queue_backtest_options(target_model, cadence)
job.modify(next_run_time=None) # Reset, then trigger immediately
from datetime import datetime, timezone
@@ -654,6 +657,8 @@ async def trigger_job(
result = {"job": job_name, "status": "triggered", "message": f"Job '{job_name}' triggered for immediate execution"}
if target_model is not None:
result["target_model"] = target_model
if cadence is not None:
result["cadence"] = cadence
return result
+1
View File
@@ -282,6 +282,7 @@ async def _qualified_setups(db: AsyncSession) -> list[dict]:
db,
live_recommendation=True,
exclude_open_trade_tickers=True,
exclude_reentry_lockdown_tickers=True,
)
config = await get_activation_config(db)
return [s for s in setups if setup_qualifies(SimpleNamespace(**s), config)]
+385 -25
View File
@@ -1,7 +1,7 @@
"""Historical backtest (Phase 1): replay the price-derived engine over stored
OHLCV and measure how the CURRENT config would have performed.
For each ticker we step through history (weekly), and at each as-of date D we
For each ticker we step through history at the selected entry cadence, and at each as-of date D we
rebuild the setup using only bars ≤ D (no lookahead), then walk the actual bars
after D to record the realized outcome. The report contains:
@@ -94,12 +94,22 @@ from app.services.scoring_service import (
compute_technical_from_arrays,
)
from app.services.sr_service import detect_gate_target_ladder, detect_sr_levels
from app.services.trade_policy import REENTRY_LOCKDOWN_SESSIONS
logger = logging.getLogger(__name__)
KEY_REPORT = "backtest_report"
STEP_DAYS = 5 # weekly cadence (≈ 5 trading days)
WEEKLY_BACKTEST_CADENCE = "weekly"
DAILY_BACKTEST_CADENCE = "daily"
DEFAULT_BACKTEST_CADENCE = WEEKLY_BACKTEST_CADENCE
BACKTEST_CADENCE_SESSIONS = {
WEEKLY_BACKTEST_CADENCE: 5,
DAILY_BACKTEST_CADENCE: 1,
}
# Compatibility alias for research scripts built around the original weekly
# replay. New code should select a cadence and call ``backtest_step_sessions``.
STEP_DAYS = BACKTEST_CADENCE_SESSIONS[WEEKLY_BACKTEST_CADENCE]
MIN_LOOKBACK = 60 # bars needed before D for indicators (EMA cross needs 51)
HORIZON = 30 # trading days to resolve an outcome (matches the evaluator)
ATR_MULTIPLIER = 1.5
@@ -156,6 +166,30 @@ def validate_backtest_target_model(value: str) -> str:
return normalized
def validate_backtest_cadence(value: str) -> str:
"""Validate the supported entry-replay cadences."""
normalized = value.strip().lower()
if normalized not in BACKTEST_CADENCE_SESSIONS:
allowed = ", ".join(BACKTEST_CADENCE_SESSIONS)
raise ValueError(
f"Unknown backtest cadence {value!r}; expected one of {allowed}"
)
return normalized
def backtest_step_sessions(cadence: str) -> int:
return BACKTEST_CADENCE_SESSIONS[validate_backtest_cadence(cadence)]
def _ranking_period(as_of: date, cadence: str) -> tuple:
"""Cross-section key for activation ranks at the selected entry cadence."""
cadence = validate_backtest_cadence(cadence)
if cadence == DAILY_BACKTEST_CADENCE:
return ("date", as_of.toordinal())
iso = as_of.isocalendar()
return ("week", iso[0], iso[1])
# ---------------------------------------------------------------------------
# RESEARCH / DIAGNOSTIC FALLBACKS (retired experiments)
#
@@ -455,14 +489,17 @@ def _replay_ticker(
activation: dict,
benchmark_closes: dict[date, float] | None = None,
target_model: str = PRODUCTION_GTL_TARGET_MODEL,
cadence: str = DEFAULT_BACKTEST_CADENCE,
) -> list[dict]:
"""Walk one ticker's history weekly, building setups and their realized outcomes."""
"""Walk one ticker at the selected cadence and resolve each setup outcome."""
cadence = validate_backtest_cadence(cadence)
step_sessions = backtest_step_sessions(cadence)
candidates: list[dict] = []
n = len(records)
if n < MIN_LOOKBACK + HORIZON:
return candidates
for i in range(MIN_LOOKBACK - 1, n - HORIZON, STEP_DAYS):
for i in range(MIN_LOOKBACK - 1, n - HORIZON, step_sessions):
window = records[: i + 1]
forward = records[i + 1 :]
forward_bars = [Bar(date=r.date, high=r.high, low=r.low) for r in forward]
@@ -511,6 +548,7 @@ def _replay_ticker(
"symbol": symbol,
"date": records[i].date.isoformat(),
"iso_week": (iso[0], iso[1]),
"ranking_period": _ranking_period(records[i].date, cadence),
"direction": s["direction"],
"entry": s["entry"],
"stop": s["stop"],
@@ -967,6 +1005,7 @@ def _replay_and_signals(
activation: dict,
benchmark_closes: dict[date, float] | None = None,
target_model: str = PRODUCTION_GTL_TARGET_MODEL,
cadence: str = DEFAULT_BACKTEST_CADENCE,
) -> tuple[list[dict], dict]:
"""The CPU-bound per-ticker work, as a top-level (picklable) function so it can
run in a worker process. Takes primitive column arrays (cheap to pickle),
@@ -986,11 +1025,81 @@ def _replay_and_signals(
activation,
benchmark_closes,
target_model,
cadence,
),
_signal_series(bars, benchmark_closes),
)
def _replay_candidates_for_period(
symbol: str,
columns: tuple,
config: dict,
activation: dict,
benchmark_closes: dict[date, float] | None,
start_date: date,
cadence: str = DEFAULT_BACKTEST_CADENCE,
) -> list[dict]:
"""Slim picklable replay used by local event studies.
Unlike the full report worker it skips factor-series construction and only
evaluates setup dates on or after ``start_date``.
"""
date_ords, opens, highs, lows, closes, volumes = columns
bars = [
SimpleNamespace(
date=date.fromordinal(o), open=op, high=hi, low=lo, close=cl, volume=vo
)
for o, op, hi, lo, cl, vo in zip(
date_ords, opens, highs, lows, closes, volumes
)
]
cadence = validate_backtest_cadence(cadence)
candidates: list[dict] = []
for i in range(
MIN_LOOKBACK - 1,
len(bars) - HORIZON,
backtest_step_sessions(cadence),
):
if bars[i].date < start_date:
continue
window = bars[: i + 1]
window_closes = [float(r.close) for r in window]
window_dates = [r.date for r in window]
residual_momentum = _residual_momentum_12_1(
window_dates,
window_closes,
len(window) - 1,
benchmark_closes,
)
vol_6m = _realized_vol_6m(window_closes, len(window) - 1)
iso = bars[i].date.isocalendar()
for setup in _window_setups(window, config, activation):
if setup["direction"] != "long":
continue
candidates.append({
"symbol": symbol,
"date": bars[i].date.isoformat(),
"iso_week": (iso[0], iso[1]),
"ranking_period": _ranking_period(bars[i].date, cadence),
"direction": "long",
"entry": setup["entry"],
"stop": setup["stop"],
"target": setup["target"],
"rr": setup["rr"],
"confidence": setup["confidence"],
"primary_prob": setup["primary_prob"],
"best_prob": setup["best_prob"],
"momentum": setup["momentum"],
"residual_momentum": residual_momentum,
"vol_6m": vol_6m,
"meets_core": setup["meets_core"],
"action": setup["action"],
"risk_level": setup["risk_level"],
})
return candidates
def _backtest_worker_count() -> int:
"""How many worker processes to replay tickers across. Capped to cpu_count-1
so a core stays free for the web server; 1 means sequential."""
@@ -1057,14 +1166,17 @@ def _assign_signal_percentiles(
value_key: str,
percentile_key: str,
) -> None:
"""Per ISO week, rank candidates by ``value_key`` and attach a 0-100
"""Per replay period, rank candidates by ``value_key`` and attach a 0-100
percentile under ``percentile_key`` (100 = strongest). Missing values get
None and therefore cannot clear a gate based on that signal."""
by_week: dict = defaultdict(list)
by_period: dict = defaultdict(list)
for c in candidates:
if c.get(value_key) is not None:
by_week[c["iso_week"]].append(c)
for group in by_week.values():
# Hand-built/research candidates predating the cadence flag retain
# the weekly key as a compatibility fallback.
period = c.get("ranking_period") or c["iso_week"]
by_period[period].append(c)
for group in by_period.values():
ordered = sorted(group, key=lambda c: c[value_key])
n = len(ordered)
for rank, c in enumerate(ordered):
@@ -1074,9 +1186,9 @@ def _assign_signal_percentiles(
def _assign_momentum_percentiles(candidates: list[dict]) -> None:
"""Per ISO week, rank candidates by their ticker's 12-1 momentum and attach a
"""Per replay period, rank candidates by 12-1 momentum and attach a
0-100 ``momentum_percentile`` (100 = highest momentum in the universe that
week). Candidates whose momentum is unknown (insufficient lookback) get None
period). Candidates whose momentum is unknown (insufficient lookback) get None
and therefore can't clear a momentum gate. Mutates ``candidates``."""
_assign_signal_percentiles(candidates, "momentum", "momentum_percentile")
@@ -1089,7 +1201,7 @@ def _assign_residual_momentum_percentiles(candidates: list[dict]) -> None:
def _assign_low_volatility_percentiles(candidates: list[dict]) -> None:
"""Per ISO week, attach volatility ranks where 100 = lowest 6-month vol."""
"""Per replay period, attach volatility ranks where 100 = lowest 6-month vol."""
_assign_signal_percentiles(candidates, "vol_6m", VOL_PERCENTILE_KEY)
for c in candidates:
raw = c.get(VOL_PERCENTILE_KEY)
@@ -1293,9 +1405,17 @@ def _simulate_portfolio(
max_positions: int = SIM_MAX_POSITIONS,
risk_per_trade: float = SIM_RISK_PER_TRADE,
atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER,
reentry_cooldown_sessions: int = 0,
initial_stop_refresh_fn: (
Callable[[str, int, float, dict, Any], float | None] | None
) = None,
post_stop_reentry_fn: (
Callable[[str, int, dict, Any], dict | None] | None
) = None,
start_date: date | None = None,
end_date: date | None = None,
include_curve: bool = False,
include_trades: bool = False,
) -> dict | None:
"""Replay the qualified setups as ONE capital-constrained book and report
portfolio economics from the daily equity curve (return, CAGR, drawdown,
@@ -1309,7 +1429,14 @@ def _simulate_portfolio(
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.
``reentry_cooldown_sessions`` blocks a ticker for that many market sessions
after an initial-stop loss. Profitable trailing-stop exits do not trigger
it. ``initial_stop_refresh_fn`` may supply a lower, point-in-time valid long
stop when the active initial stop is touched; the replacement is still
checked against the same bar. ``post_stop_reentry_fn`` turns an initial
stop-out into a stateful episode and is the only path by which that ticker
can re-enter until the callback emits a new candidate. Returns None when
there is nothing to trade.
"""
if qualified_fn is None:
def _default_qualified(c: dict) -> bool:
@@ -1362,6 +1489,14 @@ def _simulate_portfolio(
curve: list[tuple[int, float]] = []
trades: list[dict] = []
skipped_full = 0
skipped_cooldown = 0
cooldown_until_index: dict[str, int] = {}
stop_refresh_attempts = 0
stop_refreshes = 0
stop_refresh_same_bar_hits = 0
post_stop_states: dict[str, dict] = {}
post_stop_events = 0
reentry_events: list[dict] = []
technical_cache: dict[tuple[str, int], float | None] = {}
atr_cache: dict[tuple[str, int], float | None] = {}
@@ -1426,7 +1561,7 @@ def _simulate_portfolio(
atr_cache[key] = None
return atr_cache[key]
def _close_trade(sym: str, fill: float, reason: str) -> None:
def _close_trade(sym: str, fill: float, reason: str) -> dict:
nonlocal cash
pos = positions.pop(sym)
proceeds = pos["shares"] * fill
@@ -1434,16 +1569,29 @@ def _simulate_portfolio(
cash += proceeds - cost
risk = pos["entry"] - pos["initial_stop"]
trades.append({
"symbol": sym,
"entry_ord": pos["entry_ord"],
"exit_ord": o,
"entry": pos["entry"],
"initial_stop": pos["initial_stop"],
"active_stop": pos["stop"],
"fill": fill,
"pnl": proceeds - pos["shares"] * pos["entry"] - cost - pos["entry_cost"],
"r": (fill - pos["entry"]) / risk if risk > 0 else 0.0,
"hold": pos["bars_held"],
"reason": reason,
"stop_refreshes": pos["stop_refreshes"],
"is_reentry": pos["is_reentry"],
"reentry_wait_sessions": pos["reentry_wait_sessions"],
"transaction_cost": pos["entry_cost"] + cost,
})
return pos
def _marked_equity() -> float:
return cash + sum(p["shares"] * p["last_close"] for p in positions.values())
for o in calendar:
cooldown_sessions = max(0, int(reentry_cooldown_sessions))
for calendar_index, o in enumerate(calendar):
# 1) exits on today's bars (stop intraday, target intraday, time at close)
for sym in list(positions):
pos = positions[sym]
@@ -1460,7 +1608,41 @@ def _simulate_portfolio(
if pos["stop"] > pos["initial_stop"] + 1e-9
else "stop"
)
_close_trade(sym, min(pos["stop"], bar.open), reason)
survived_refresh = False
if reason == "stop" and initial_stop_refresh_fn is not None:
stop_refresh_attempts += 1
refreshed_stop = initial_stop_refresh_fn(
sym, o, float(pos["stop"]), pos, bar
)
if (
refreshed_stop is not None
and 0 < float(refreshed_stop) < pos["stop"] - 1e-9
):
pos["stop"] = float(refreshed_stop)
pos["stop_refreshes"] += 1
stop_refreshes += 1
if bar.low > pos["stop"]:
survived_refresh = True
else:
stop_refresh_same_bar_hits += 1
if not survived_refresh:
fill = min(pos["stop"], bar.open)
closed_pos = _close_trade(sym, fill, reason)
if reason == "stop" and cooldown_sessions:
cooldown_until_index[sym] = calendar_index + cooldown_sessions
if reason == "stop" and post_stop_reentry_fn is not None:
post_stop_events += 1
post_stop_states[sym] = {
"stop_ord": o,
"stop_calendar_index": calendar_index,
"stop_day_high": float(bar.high),
"stop_day_low": float(bar.low),
"stop_day_close": float(bar.close),
"exit_fill": float(fill),
"previous_entry": float(closed_pos["entry"]),
"previous_stop": float(closed_pos["initial_stop"]),
"gate_went_unqualified": False,
}
continue
if exit_policy in ("target", "atr_trail3_target") and pos["target"] and bar.high >= pos["target"]:
_close_trade(sym, pos["target"], "target")
@@ -1493,8 +1675,29 @@ def _simulate_portfolio(
# 2) entries at today's close, best momentum first
equity = _marked_equity()
fixed_todays = list(entries_by_ord.get(o, ()))
reentry_todays: list[dict] = []
if post_stop_reentry_fn is not None:
fixed_todays = [
candidate
for candidate in fixed_todays
if candidate["symbol"] not in post_stop_states
]
for sym, state in list(post_stop_states.items()):
bar = _bar(sym, o)
if bar is None:
continue
state["sessions_since_stop"] = (
calendar_index - state["stop_calendar_index"]
)
candidate = post_stop_reentry_fn(sym, o, state, bar)
if candidate is None:
continue
tagged = dict(candidate)
tagged["_post_stop_reentry"] = True
reentry_todays.append(tagged)
todays = sorted(
entries_by_ord.get(o, ()),
fixed_todays + reentry_todays,
key=lambda c: c.get(ranking_key) or 0.0,
reverse=True,
)
@@ -1502,6 +1705,9 @@ def _simulate_portfolio(
sym = c["symbol"]
if sym in positions:
continue
if calendar_index < cooldown_until_index.get(sym, -1):
skipped_cooldown += 1
continue
if len(positions) >= max_positions:
skipped_full += 1
continue
@@ -1518,9 +1724,23 @@ def _simulate_portfolio(
continue
entry_cost = shares * entry * COST_PER_SIDE
cash -= shares * entry + entry_cost
is_reentry = bool(c.get("_post_stop_reentry"))
reentry_wait_sessions: int | None = None
if is_reentry:
state = post_stop_states.pop(sym, None)
if state is not None:
reentry_wait_sessions = int(state["sessions_since_stop"])
reentry_events.append({
"symbol": sym,
"stop_ord": state["stop_ord"],
"reentry_ord": o,
"wait_sessions": reentry_wait_sessions,
"reason": c.get("_reentry_reason"),
})
positions[sym] = {
"shares": shares,
"entry": entry,
"entry_ord": o,
"initial_stop": stop,
"stop": stop,
"target": float(c["target"]) if c.get("target") else None,
@@ -1528,6 +1748,9 @@ def _simulate_portfolio(
"bars_held": 0,
"last_close": entry,
"highest_close": entry,
"stop_refreshes": 0,
"is_reentry": is_reentry,
"reentry_wait_sessions": reentry_wait_sessions,
}
equity = _marked_equity()
@@ -1659,6 +1882,42 @@ def _simulate_portfolio(
result["equity_curve"] = curve_payload
if benchmark_payload is not None:
result["benchmark_curve"] = benchmark_payload
if cooldown_sessions:
result["reentry_cooldown_sessions"] = cooldown_sessions
result["skipped_cooldown"] = skipped_cooldown
if initial_stop_refresh_fn is not None:
result["stop_refresh_attempts"] = stop_refresh_attempts
result["stop_refreshes"] = stop_refreshes
result["stop_refresh_same_bar_hits"] = stop_refresh_same_bar_hits
if post_stop_reentry_fn is not None:
result["post_stop_events"] = post_stop_events
result["post_stop_reentries"] = len(reentry_events)
result["post_stop_states_open_at_end"] = len(post_stop_states)
result["reentry_events"] = [
{
**{
key: value
for key, value in event.items()
if key not in {"stop_ord", "reentry_ord"}
},
"stop_date": date.fromordinal(event["stop_ord"]).isoformat(),
"reentry_date": date.fromordinal(event["reentry_ord"]).isoformat(),
}
for event in reentry_events
]
if include_trades:
result["trade_details"] = [
{
**{
key: value
for key, value in trade.items()
if key not in {"entry_ord", "exit_ord"}
},
"entry_date": date.fromordinal(trade["entry_ord"]).isoformat(),
"exit_date": date.fromordinal(trade["exit_ord"]).isoformat(),
}
for trade in trades
]
return result
@@ -2018,19 +2277,35 @@ PORTFOLIO_MONITOR_STRATEGIES: tuple[dict, ...] = (
"exit_policy": "hold",
},
{
"strategy": PRODUCTION_PORTFOLIO_STRATEGY,
"label": "Production: residual/high-vol 80/20 + 3x ATR trail",
"strategy": "production_live_no_lockdown",
"label": "Live setup + 3x ATR trail (no re-entry lockdown)",
"description": (
"The live strategy: production activation gate and Admin exit policy "
"as currently configured, 80/20 residual/high-vol rank."
"Exact live activation, ordering, and Admin exit policy, with only "
"the post-stop re-entry lockdown disabled as the comparison baseline."
),
"entry_variant": "residual80_highvol_blend80_20_fixed10",
"exit_policy": "atr_trail3",
"reentry_lockdown_sessions": 0,
"use_live_config": True,
"comparison_arm": "live_no_lockdown",
},
{
"strategy": PRODUCTION_PORTFOLIO_STRATEGY,
"label": "Production: residual/high-vol 80/20 + 3x ATR trail + 5-session lockdown",
"description": (
"The live strategy: production activation gate and Admin exit policy "
"as currently configured, 80/20 residual/high-vol rank, and a "
"five-session re-entry lockdown after an initial-stop exit."
),
"entry_variant": "residual80_highvol_blend80_20_fixed10",
"exit_policy": "atr_trail3",
"reentry_lockdown_sessions": REENTRY_LOCKDOWN_SESSIONS,
# The production row replays what the platform actually does right now:
# the live qualification flag (runtime Admin activation settings) and the
# live Admin exit policy, instead of the frozen research-variant gate.
"use_live_config": True,
"is_production": True,
"comparison_arm": "live_lockdown_5",
},
)
@@ -2149,6 +2424,9 @@ def _min_rr_sweep(
exit_policy = str(strategy["exit_policy"])
row_hold_days = hold_days
trail_multiplier = ATR_TRAIL_MULTIPLIER
reentry_lockdown_sessions = int(
strategy.get("reentry_lockdown_sessions", 0)
)
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"
@@ -2188,6 +2466,7 @@ def _min_rr_sweep(
max_positions=int(entry_cfg["max_positions"]),
risk_per_trade=float(entry_cfg["risk_per_trade"]),
atr_trail_multiplier=trail_multiplier,
reentry_cooldown_sessions=reentry_lockdown_sessions,
start_date=sweep_start,
)
if sim is None:
@@ -2212,6 +2491,7 @@ def _min_rr_sweep(
"live_qualified_setups": live_qualified,
"reproduces_production_gate": reproduces,
"exit_policy": exit_policy,
"reentry_lockdown_sessions": reentry_lockdown_sessions,
"entries_from": sweep_start.isoformat() if sweep_start else None,
"window": "out-of-sample (test)" if sweep_start else "full history (in-sample)",
"rows": rows,
@@ -2267,6 +2547,9 @@ def _holdout_evaluation(
exit_policy = str(strategy["exit_policy"])
row_hold_days = hold_days
trail_multiplier = ATR_TRAIL_MULTIPLIER
reentry_lockdown_sessions = int(
strategy.get("reentry_lockdown_sessions", 0)
)
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"
@@ -2296,6 +2579,7 @@ def _holdout_evaluation(
max_positions=int(entry_cfg["max_positions"]),
risk_per_trade=float(entry_cfg["risk_per_trade"]),
atr_trail_multiplier=trail_multiplier,
reentry_cooldown_sessions=reentry_lockdown_sessions,
start_date=start,
end_date=end,
include_curve=True,
@@ -2307,6 +2591,7 @@ def _holdout_evaluation(
return {
"split_date": split.isoformat(),
"strategy": strategy["strategy"],
"reentry_lockdown_sessions": reentry_lockdown_sessions,
"rows": rows,
"note": (
"Train = entries before the split; test = entries on/after it. The two "
@@ -2339,6 +2624,9 @@ def _portfolio_monitor(
# policy. The overlay opts into this deliberately so only ordering
# changes relative to the production row.
use_live = bool(strategy.get("use_live_config"))
reentry_lockdown_sessions = int(
strategy.get("reentry_lockdown_sessions", 0)
)
exit_policy = str(strategy["exit_policy"])
row_hold_days = hold_days
trail_multiplier = ATR_TRAIL_MULTIPLIER
@@ -2367,6 +2655,7 @@ def _portfolio_monitor(
max_positions=int(entry_cfg["max_positions"]),
risk_per_trade=float(entry_cfg["risk_per_trade"]),
atr_trail_multiplier=trail_multiplier,
reentry_cooldown_sessions=reentry_lockdown_sessions,
start_date=start,
include_curve=True,
)
@@ -2377,10 +2666,12 @@ def _portfolio_monitor(
"label": strategy["label"],
"description": strategy["description"],
"is_production": bool(strategy.get("is_production")),
"comparison_arm": strategy.get("comparison_arm"),
"entry_variant": strategy["entry_variant"],
"ranking_key": ranking_key,
"exit_policy": exit_policy,
"live_exit_mode": live_exit_mode,
"reentry_lockdown_sessions": reentry_lockdown_sessions,
"lookback": lookback["lookback"],
"lookback_label": lookback["label"],
**sim,
@@ -2393,6 +2684,10 @@ def _portfolio_monitor(
"label": s["label"],
"description": s["description"],
"is_production": bool(s.get("is_production")),
"comparison_arm": s.get("comparison_arm"),
"reentry_lockdown_sessions": int(
s.get("reentry_lockdown_sessions", 0)
),
}
for s in strategies
],
@@ -2405,7 +2700,48 @@ def _portfolio_monitor(
"Portfolio monitor runs supported named strategies across cached lookbacks. "
"The structural overlay appears only in its explicit research arm and changes "
"ordering, not production qualification. Local snapshot backtests remain the "
"research surface for broad variant sweeps."
"research surface for broad variant sweeps. The production row applies the "
"same five-session post-initial-stop re-entry lockdown as the live setup list."
),
}
def _production_cadence_comparison(
monitor: dict | None,
cadence: str,
) -> dict | None:
"""Compact full-history live/no-lockdown vs live/5-session comparison."""
if not monitor:
return None
arms: list[dict] = []
for row in monitor.get("runs") or []:
comparison_arm = row.get("comparison_arm")
if not comparison_arm or row.get("lookback") != "all":
continue
compact = {
key: value
for key, value in row.items()
if key not in {"equity_curve", "benchmark_curve"}
}
arm_name = (
"prod_live_setup"
if comparison_arm == "live_no_lockdown"
else "cooldown_5"
)
compact["arm"] = f"{arm_name}_{cadence}"
compact["entry_cadence"] = cadence
arms.append(compact)
if not arms:
return None
arms.sort(key=lambda row: int(row.get("reentry_lockdown_sessions", 0)))
return {
"entry_cadence": cadence,
"lookback": "all",
"arms": arms,
"note": (
"Both arms use the exact same live gate, ordering, Admin exit policy, "
"fees, and candidate cadence. Only the five-session post-stop "
"re-entry lockdown changes."
),
}
@@ -2618,7 +2954,8 @@ def _build_recommendation(report: dict) -> dict:
if production_row is not None:
headline = (
"Production baseline: residual/high-vol 80/20 entry rank with a "
"3x ATR trailing exit and 30-trading-day max hold."
"3x ATR trailing exit, 30-trading-day max hold, and 5-session "
"re-entry lockdown after an initial stop."
)
if (
production_row.get("cagr_pct") is not None
@@ -2787,9 +3124,11 @@ async def run_backtest(
progress_cb: Callable[[int, int, str], None] | None = None,
*,
target_model: str = PRODUCTION_GTL_TARGET_MODEL,
cadence: str = DEFAULT_BACKTEST_CADENCE,
) -> dict:
"""Replay every ticker and aggregate the Phase-1 reports for the current config."""
target_model = validate_backtest_target_model(target_model)
cadence = validate_backtest_cadence(cadence)
config = await get_recommendation_config(db)
activation = await get_activation_config(db)
@@ -2798,7 +3137,9 @@ async def run_backtest(
total = len(tickers)
candidates: list[dict] = []
# collected[signal_name][iso_week] -> list of (signal_value, forward_return)
# Signal IC remains a weekly, non-overlapping diagnostic regardless of the
# entry cadence. Production activation ranks are assigned from candidates
# at their own weekly or exact-date ``ranking_period`` below.
collected: dict = defaultdict(lambda: defaultdict(list))
# Residual momentum needs a point-in-time benchmark return stream. Best-effort:
@@ -2855,6 +3196,7 @@ async def run_backtest(
pool, _replay_and_signals, ticker.symbol, columns, config, activation,
benchmark_closes,
target_model,
cadence,
))
for result in await asyncio.gather(*futures, return_exceptions=True):
if isinstance(result, Exception):
@@ -2877,6 +3219,7 @@ async def run_backtest(
_replay_and_signals, ticker.symbol, columns, config, activation,
benchmark_closes,
target_model,
cadence,
))
except Exception:
logger.exception("Backtest replay failed for %s", ticker.symbol)
@@ -2987,13 +3330,19 @@ async def run_backtest(
"candidates": len(candidates),
"qualified": len(qualified),
"params": {
"step_days": STEP_DAYS,
# Keep step_days for old report consumers; the value counts stored
# market sessions rather than calendar days.
"step_days": backtest_step_sessions(cadence),
"step_sessions": backtest_step_sessions(cadence),
"entry_cadence": cadence,
"signal_eval_cadence": WEEKLY_BACKTEST_CADENCE,
"horizon_days": HORIZON,
"min_lookback": MIN_LOOKBACK,
"cost_per_side_pct": round(COST_PER_SIDE * 100, 3),
"target_model": target_model,
"target_model_label": BACKTEST_TARGET_MODELS[target_model],
"is_production_target_model": target_model == PRODUCTION_GTL_TARGET_MODEL,
"production_reentry_lockdown_sessions": REENTRY_LOCKDOWN_SESSIONS,
},
"activation": activation,
"overall_qualified": _bucket_stats(qualified),
@@ -3055,6 +3404,11 @@ async def run_backtest(
),
},
"portfolio_monitor": portfolio_monitor_report,
"production_cadence_comparison": (
_production_cadence_comparison(portfolio_monitor_report, cadence)
if target_model == PRODUCTION_GTL_TARGET_MODEL
else None
),
"holdout": holdout_report,
"min_rr_sweep": min_rr_sweep_report,
"target_model_diagnostics": _target_model_diagnostics(
@@ -3090,9 +3444,15 @@ async def run_and_store(
progress_cb: Callable[[int, int, str], None] | None = None,
*,
target_model: str = PRODUCTION_GTL_TARGET_MODEL,
cadence: str = DEFAULT_BACKTEST_CADENCE,
) -> dict:
"""Run the backtest and cache the report in a SystemSetting. Job entrypoint."""
report = await run_backtest(db, progress_cb, target_model=target_model)
report = await run_backtest(
db,
progress_cb,
target_model=target_model,
cadence=cadence,
)
await update_setting(db, KEY_REPORT, json.dumps(report))
return report
+8
View File
@@ -20,6 +20,7 @@ from app.services.outcome_service import (
Bar,
evaluate_setup_against_bars,
)
from app.services.trade_policy import get_reentry_lockdowns
# Exit policy for OPEN paper trades (auto-close). Production defaults to the
# July 2026 promoted strategy: initial stop + 3x ATR trailing stop, with a max
@@ -318,6 +319,13 @@ async def create_trade(
raise ValidationError("shares and entry_price must be positive")
ticker = await _get_ticker(db, symbol)
remaining_sessions = (await get_reentry_lockdowns(db)).get(ticker.id)
if remaining_sessions is not None:
suffix = "session" if remaining_sessions == 1 else "sessions"
raise ValidationError(
f"{ticker.symbol} is in a post-stop re-entry lockdown: "
f"{remaining_sessions} market {suffix} remaining"
)
trade = PaperTrade(
user_id=user_id,
ticker_id=ticker.id,
+23 -3
View File
@@ -29,6 +29,7 @@ from app.models.trade_setup import TradeSetup
from app.services.indicator_service import _extract_ohlcv, compute_atr
from app.services.price_service import query_ohlcv
from app.services.sr_service import detect_gate_target_ladder
from app.services.trade_policy import get_reentry_lockdowns
from app.services.recommendation_service import (
_risk_level_from_conflicts,
build_recommendation_snapshot,
@@ -771,6 +772,8 @@ async def get_trade_setups(
symbol: str | None = None,
live_recommendation: bool = False,
exclude_open_trade_tickers: bool = False,
exclude_reentry_lockdown_tickers: bool = False,
include_reentry_lockdown: bool = False,
) -> list[dict]:
"""Get latest stored trade setups, optionally filtered.
@@ -794,15 +797,23 @@ async def get_trade_setups(
stmt = stmt.where(TradeSetup.confidence_score >= min_confidence)
if recommended_action is not None and not live_recommendation:
stmt = stmt.where(TradeSetup.recommended_action == recommended_action)
excluded_ticker_ids: set[int] = set()
reentry_lockdowns: dict[int, int] = {}
if exclude_open_trade_tickers:
open_trade_result = await db.execute(
select(PaperTrade.ticker_id)
.where(PaperTrade.status == "open")
.distinct()
)
open_ticker_ids = {ticker_id for ticker_id, in open_trade_result.all()}
if open_ticker_ids:
stmt = stmt.where(~TradeSetup.ticker_id.in_(open_ticker_ids))
excluded_ticker_ids.update(
ticker_id for ticker_id, in open_trade_result.all()
)
if exclude_reentry_lockdown_tickers or include_reentry_lockdown:
reentry_lockdowns = await get_reentry_lockdowns(db)
if exclude_reentry_lockdown_tickers:
excluded_ticker_ids.update(reentry_lockdowns)
if excluded_ticker_ids:
stmt = stmt.where(~TradeSetup.ticker_id.in_(excluded_ticker_ids))
stmt = stmt.order_by(TradeSetup.detected_at.desc(), TradeSetup.id.desc())
@@ -855,6 +866,15 @@ async def get_trade_setups(
),
reverse=True,
)
if include_reentry_lockdown:
ticker_by_setup_id = {
setup.id: setup.ticker_id for setup, _ in latest_rows
}
for row in rows_out:
ticker_id = ticker_by_setup_id.get(row["id"])
row["reentry_lockdown_remaining_sessions"] = (
reentry_lockdowns.get(ticker_id) if ticker_id is not None else None
)
return rows_out
+120
View File
@@ -0,0 +1,120 @@
"""Shared live/backtest trading-policy constants and availability checks."""
from __future__ import annotations
from collections import defaultdict
from datetime import date, datetime, timezone
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.benchmark_price import BenchmarkPrice
from app.models.ohlcv import OHLCVRecord
from app.models.paper_trade import PaperTrade
from app.services.benchmark_service import BENCHMARK_SYMBOL
# A ticker stopped at its initial stop may qualify again immediately, but the
# July 2026 event study showed that waiting five market sessions materially
# improved the production book. The stop session is wait_session=0; the first
# permitted re-entry is wait_session=5, provided the normal gate still passes.
REENTRY_LOCKDOWN_SESSIONS = 5
async def get_reentry_lockdowns(
db: AsyncSession,
*,
as_of: date | None = None,
sessions: int = REENTRY_LOCKDOWN_SESSIONS,
) -> dict[int, int]:
"""Return ``{ticker_id: remaining_sessions}`` for active lockdowns.
SPY is the canonical calendar for the platform's US-equity universe. When
the stored benchmark history does not reach an older stop, only that
ticker's own OHLCV dates are used as a conservative fallback. Unrelated
ticker dates can therefore never shorten a lockdown.
"""
sessions = max(0, int(sessions))
if sessions == 0:
return {}
session_cutoff = as_of or datetime.now(timezone.utc).date()
stop_result = await db.execute(
select(
PaperTrade.ticker_id,
func.max(PaperTrade.closed_at).label("last_stop_at"),
)
.where(
PaperTrade.status == "closed",
PaperTrade.close_reason == "stop",
PaperTrade.closed_at.is_not(None),
)
.group_by(PaperTrade.ticker_id)
)
stop_dates = {
ticker_id: stopped_at.date()
for ticker_id, stopped_at in stop_result.all()
if stopped_at is not None and stopped_at.date() <= session_cutoff
}
if not stop_dates:
return {}
benchmark_result = await db.execute(
select(BenchmarkPrice.date)
.where(
BenchmarkPrice.symbol == BENCHMARK_SYMBOL,
BenchmarkPrice.date <= session_cutoff,
)
.order_by(BenchmarkPrice.date.asc())
)
benchmark_dates = [row[0] for row in benchmark_result.all()]
lockdowns: dict[int, int] = {}
fallback_stops: dict[int, date] = {}
first_benchmark_date = benchmark_dates[0] if benchmark_dates else None
for ticker_id, stop_date in stop_dates.items():
completed = sum(day > stop_date for day in benchmark_dates)
if completed >= sessions:
continue
if first_benchmark_date is not None and first_benchmark_date <= stop_date:
lockdowns[ticker_id] = sessions - completed
else:
# The benchmark table starts after this stop (or is empty), so it
# cannot prove how many sessions elapsed. Resolve only this ticker
# against its own bars instead of using universe-wide dates.
fallback_stops[ticker_id] = stop_date
if fallback_stops:
own_session_result = await db.execute(
select(OHLCVRecord.ticker_id, OHLCVRecord.date)
.where(
OHLCVRecord.ticker_id.in_(fallback_stops),
OHLCVRecord.date > min(fallback_stops.values()),
OHLCVRecord.date <= session_cutoff,
)
.distinct()
)
own_dates: dict[int, set[date]] = defaultdict(set)
for ticker_id, market_date in own_session_result.all():
own_dates[ticker_id].add(market_date)
for ticker_id, stop_date in fallback_stops.items():
completed = sum(day > stop_date for day in own_dates[ticker_id])
if completed < sessions:
lockdowns[ticker_id] = sessions - completed
return lockdowns
async def get_reentry_lockdown_ticker_ids(
db: AsyncSession,
*,
as_of: date | None = None,
sessions: int = REENTRY_LOCKDOWN_SESSIONS,
) -> set[int]:
"""Compatibility wrapper for callers that only need blocked ticker ids."""
return set(
await get_reentry_lockdowns(
db,
as_of=as_of,
sessions=sessions,
)
)
+6 -1
View File
@@ -201,9 +201,11 @@ export interface TriggerJobResponse {
status: 'triggered' | 'busy' | 'blocked' | 'not_found';
message: string;
target_model?: BacktestTargetModel;
cadence?: BacktestCadence;
}
export type BacktestTargetModel = 'production_gtl' | 'structural_sr';
export type BacktestCadence = 'weekly' | 'daily';
export function listJobs() {
return apiClient.get<JobStatus[]>('admin/jobs').then((r) => r.data);
@@ -219,7 +221,10 @@ export function toggleJob(jobName: string, enabled: boolean) {
.then((r) => r.data);
}
export function triggerJob(jobName: string, options?: { target_model?: BacktestTargetModel }) {
export function triggerJob(
jobName: string,
options?: { target_model?: BacktestTargetModel; cadence?: BacktestCadence },
) {
return apiClient
.post<TriggerJobResponse>(`admin/jobs/${jobName}/trigger`, options)
.then((r) => r.data);
@@ -2,7 +2,7 @@ import { useMemo, useState } from 'react';
import { useMutation, useQueryClient } from '@tanstack/react-query';
import { useBacktestReport } from '../../hooks/useMarketRegime';
import { triggerJob } from '../../api/admin';
import type { BacktestTargetModel } from '../../api/admin';
import type { BacktestCadence, BacktestTargetModel } from '../../api/admin';
import { Button } from '../ui/Button';
import { Callout } from '../ui/Callout';
import { Disclosure } from '../ui/Disclosure';
@@ -145,6 +145,7 @@ export function BacktestPanel() {
const [selectedStrategy, setSelectedStrategy] = useState('');
const [selectedLookback, setSelectedLookback] = useState('');
const [targetModel, setTargetModel] = useState<BacktestTargetModel>('production_gtl');
const [cadence, setCadence] = useState<BacktestCadence>('weekly');
const monitor = report?.portfolio_monitor ?? null;
const activeStrategy =
@@ -161,11 +162,11 @@ export function BacktestPanel() {
);
const run = useMutation({
mutationFn: () => triggerJob('backtest', { target_model: targetModel }),
mutationFn: () => triggerJob('backtest', { target_model: targetModel, cadence }),
onSuccess: (res) => {
if (res.status === 'triggered') {
const label = targetModel === 'production_gtl' ? 'Live GTL' : 'Structural S/R comparison';
toast.addToast('success', `${label} backtest started — results appear when it finishes.`);
toast.addToast('success', `${label} ${cadence} backtest started — results appear when it finishes.`);
setTimeout(() => queryClient.invalidateQueries({ queryKey: ['backtest-report'] }), 8000);
} else {
toast.addToast('info', res.message || 'Could not start backtest');
@@ -180,7 +181,7 @@ export function BacktestPanel() {
<div className="flex flex-wrap items-start justify-between gap-3">
<Disclosure summary="How this is measured">
<p className="max-w-2xl text-xs text-gray-400">
The backtest replays the current config weekly through history at each point the setup is
The backtest replays the current config at the selected cadence at each point the setup is
rebuilt using only data up to that day (no lookahead) and the following ~30 trading days decide
its outcome then simulates one capital-constrained book against the S&P 500. Sentiment and
fundamentals are held neutral (no point-in-time history). ~6 months is roughly one market regime,
@@ -238,6 +239,56 @@ export function BacktestPanel() {
</span>
</label>
</fieldset>
<fieldset className="grid w-full grid-cols-2 gap-2 sm:w-[34rem]">
<legend className="mb-1 text-[11px] font-medium uppercase tracking-wider text-gray-500">
Entry cadence
</legend>
<label
className={`cursor-pointer rounded-lg border px-3 py-2 transition-colors focus-within:ring-2 focus-within:ring-blue-400/60 ${
cadence === 'weekly'
? 'border-blue-400/60 bg-blue-500/10'
: 'border-white/10 bg-white/[0.03] hover:border-white/20'
}`}
>
<input
className="sr-only"
type="radio"
name="backtest-cadence"
value="weekly"
checked={cadence === 'weekly'}
onChange={() => setCadence('weekly')}
/>
<span className="flex items-center justify-between gap-2 text-sm font-medium text-gray-100">
Weekly
<span className="rounded-full border border-blue-400/40 bg-blue-400/10 px-2 py-0.5 text-[9px] font-semibold uppercase tracking-widest text-blue-300">
Default
</span>
</span>
<span className="mt-1 block text-[11px] leading-4 text-gray-500">
Resource-safe server run at five-session intervals.
</span>
</label>
<label
className={`cursor-pointer rounded-lg border px-3 py-2 transition-colors focus-within:ring-2 focus-within:ring-amber-400/60 ${
cadence === 'daily'
? 'border-amber-400/50 bg-amber-500/10'
: 'border-white/10 bg-white/[0.03] hover:border-white/20'
}`}
>
<input
className="sr-only"
type="radio"
name="backtest-cadence"
value="daily"
checked={cadence === 'daily'}
onChange={() => setCadence('daily')}
/>
<span className="text-sm font-medium text-gray-200">Daily</span>
<span className="mt-1 block text-[11px] leading-4 text-amber-300/80">
Research run: roughly 5× the replay work; prefer the offline snapshot runner.
</span>
</label>
</fieldset>
<Button onClick={() => run.mutate()} loading={run.isPending} className="shrink-0">
{run.isPending ? 'Starting…' : report ? 'Re-run backtest' : 'Run backtest'}
</Button>
@@ -257,7 +308,8 @@ export function BacktestPanel() {
<>
<p className="text-[11px] text-gray-500">
Ran {timeAgo(report.generated_at)} · {report.tickers} tickers · {report.candidates} setups
({report.qualified} qualified) · weekly cadence, {report.params.horizon_days}-day horizon
({report.qualified} qualified) · {report.params.entry_cadence ?? 'weekly'} cadence,
{' '}{report.params.horizon_days}-day horizon
{report.params.cost_per_side_pct != null && (
<> · net of {report.params.cost_per_side_pct}%/side costs</>
)}
@@ -321,6 +373,9 @@ export function BacktestPanel() {
<p className="text-[11px] text-gray-500">
Avg hold {fmtDays(monitorRun.avg_hold_days)} · Best {fmtR(monitorRun.best_trade_r)} / Worst{' '}
{fmtR(monitorRun.worst_trade_r)} · Avg P&amp;L per trade {fmtMoney(monitorRun.avg_trade_pnl)}
{monitorRun.reentry_lockdown_sessions ? (
<> · Re-entry lockdown {monitorRun.reentry_lockdown_sessions} market sessions after initial stop</>
) : null}
</p>
{monitorRun.yearly_returns && monitorRun.yearly_returns.length > 0 && (
@@ -66,14 +66,19 @@ function entryDrift(setup: TradeSetup, currentPrice?: number) {
return { pct, r, status };
}
/**
* The only state with no tradeable setup left: price has gone through the stop.
* Returns null when there's no live price.
*/
type NotActionableState =
| { kind: 'lockdown'; remainingSessions: number }
| { kind: 'invalidated' }
| null;
function notActionableState(setup: TradeSetup, currentPrice?: number) {
const remainingSessions = setup.reentry_lockdown_remaining_sessions ?? 0;
if (remainingSessions > 0) {
return { kind: 'lockdown', remainingSessions } satisfies NotActionableState;
}
if (currentPrice == null) return null;
if (entryDrift(setup, currentPrice)?.status !== 'invalidated') return null;
return { invalidated: true };
return { kind: 'invalidated' } satisfies NotActionableState;
}
function riskClass(risk: TradeSetup['risk_level']) {
@@ -218,9 +223,6 @@ function SetupCard({ setup, action, currentPrice, risk, regime, exitPolicy, sele
const exitPlan = deriveExitPlan(setup, exitPolicy);
const honorsTarget = exitPlan?.honorsTarget ?? false;
// Only price through the stop leaves no tradeable setup.
const notActionable = notActionableState(setup, currentPrice) != null;
const createTrade = useCreatePaperTrade();
const [taking, setTaking] = useState(false);
const [takeShares, setTakeShares] = useState<number>(sizing?.shares ?? 0);
@@ -268,7 +270,27 @@ function SetupCard({ setup, action, currentPrice, risk, regime, exitPolicy, sele
);
};
if (notActionable) {
const inactiveState = notActionableState(setup, currentPrice);
if (inactiveState?.kind === 'lockdown') {
const remaining = inactiveState.remainingSessions;
return (
<div data-direction={setup.direction} className="rounded-xl border border-amber-400/20 bg-amber-400/[0.04] p-4">
<div className="flex flex-wrap items-center gap-2">
<DirTag direction={setup.direction} />
<span className="num text-[10px] uppercase tracking-[0.16em] text-amber-300">post-stop lockdown</span>
<span className="num ml-auto text-xs text-gray-500">
{remaining} market session{remaining === 1 ? '' : 's'} remaining
</span>
</div>
<p className="mt-2 text-[11.5px] leading-relaxed text-gray-400">
This setup remains visible for context but cannot be marked as taken. Once the lockdown expires,
the scanner recalculates the normal gate before it can become actionable again.
</p>
</div>
);
}
if (inactiveState?.kind === 'invalidated') {
const dir = setup.direction.toUpperCase();
return (
<div data-direction={setup.direction} className="rounded-xl border border-white/[0.07] p-4">
@@ -617,7 +639,21 @@ export function RecommendationPanel({ symbol, longSetup, shortSetup, currentPric
<div className="min-w-0">
{preferredInactive ? (
<span className="text-sm font-semibold text-gray-400">
No current setup <span className="font-normal text-gray-500">(last {preferredDirection} bias {recommendationActionLabel(action).toLowerCase()} invalidated at the stop)</span>
{preferredInactive.kind === 'lockdown' ? (
<>
Re-entry paused{' '}
<span className="font-normal text-gray-500">
({preferredInactive.remainingSessions} market session{preferredInactive.remainingSessions === 1 ? '' : 's'} remaining after stop)
</span>
</>
) : (
<>
No current setup{' '}
<span className="font-normal text-gray-500">
(last {preferredDirection} bias {recommendationActionLabel(action).toLowerCase()} invalidated at the stop)
</span>
</>
)}
</span>
) : (() => {
const reasoning = summary?.reasoning ?? '';
+5
View File
@@ -43,6 +43,7 @@ export function liveRiskReward(setup: TradeSetup, currentPrice: number): number
* app/services/qualification.py keep the two in sync.
*/
export function qualifiesSetup(setup: TradeSetup, config: ActivationConfig): boolean {
if ((setup.reentry_lockdown_remaining_sessions ?? 0) > 0) return false;
if (setup.rr_ratio < config.min_rr) return false;
// Live R:R from current price — drops setups whose price has already run
// toward target (reward consumed) or through the stop.
@@ -79,6 +80,10 @@ export function qualifiesSetup(setup: TradeSetup, config: ActivationConfig): boo
* qualifiesSetup rule-for-rule (keep the order in sync).
*/
export function disqualifyReason(setup: TradeSetup, config: ActivationConfig): string | null {
const lockdownRemaining = setup.reentry_lockdown_remaining_sessions ?? 0;
if (lockdownRemaining > 0) {
return `post-stop lockdown · ${lockdownRemaining} session${lockdownRemaining === 1 ? '' : 's'} remaining`;
}
if (setup.rr_ratio < config.min_rr) {
return `R:R ${setup.rr_ratio.toFixed(1)} below gate ${config.min_rr.toFixed(1)}`;
}
+15 -1
View File
@@ -144,6 +144,7 @@ export interface TradeSetup {
momentum_percentile?: number | null;
strategy_rank?: number | null;
volatility_percentile?: number | null;
reentry_lockdown_remaining_sessions?: number | null;
context_as_of?: TradeSetupContextAsOf | null;
recommendation_summary?: RecommendationSummary;
}
@@ -355,15 +356,24 @@ export interface BacktestPortfolioMonitorRun extends BacktestPortfolioPolicy {
label: string;
description: string;
is_production: boolean;
comparison_arm?: 'live_no_lockdown' | 'live_lockdown_5' | null;
entry_variant: string;
exit_policy: string;
reentry_lockdown_sessions?: number;
lookback: string;
lookback_label: string;
}
export interface BacktestPortfolioMonitor {
production_strategy: string;
strategies: { strategy: string; label: string; description: string; is_production: boolean }[];
strategies: {
strategy: string;
label: string;
description: string;
is_production: boolean;
comparison_arm?: 'live_no_lockdown' | 'live_lockdown_5' | null;
reentry_lockdown_sessions?: number;
}[];
lookbacks: { lookback: string; label: string }[];
runs: BacktestPortfolioMonitorRun[];
note?: string;
@@ -396,12 +406,16 @@ export interface BacktestReport {
qualified: number;
params: {
step_days: number;
step_sessions?: number;
entry_cadence?: 'weekly' | 'daily';
signal_eval_cadence?: 'weekly';
horizon_days: number;
min_lookback: number;
cost_per_side_pct?: number;
target_model?: 'production_gtl' | 'structural_sr';
target_model_label?: string;
is_production_target_model?: boolean;
production_reentry_lockdown_sessions?: number;
};
overall_qualified: BacktestBucket;
overall_all: BacktestBucket;
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+179
View File
@@ -0,0 +1,179 @@
"""Run the four production cadence/lockdown arms on one offline snapshot.
The command executes the complete backtest once weekly and once daily. Each
backtest contains two otherwise identical live-policy portfolio arms: no
post-stop lockdown and the production five-session lockdown. It writes both
full reports plus one compact four-arm comparison report.
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
def _parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"snapshot",
help="SQLite snapshot created by scripts/create_backtest_snapshot.py.",
)
parser.add_argument(
"--out-dir",
default="reports",
help="Directory for the weekly, daily, and comparison JSON reports.",
)
parser.add_argument(
"--prefix",
default=None,
help="Output prefix. Defaults to backtest-cadence-<timestamp>.",
)
parser.add_argument(
"--workers",
type=int,
default=None,
help="Override worker count; on a powerful offline PC use CPU count minus one.",
)
parser.add_argument(
"--allow-spawn",
action="store_true",
help="Enable multiprocessing spawn for the offline Windows run.",
)
parser.add_argument("--quiet", action="store_true", help="Hide ticker progress.")
return parser.parse_args()
def _sqlite_url(path: Path) -> str:
return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
def _write_json(path: Path, payload: dict) -> None:
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
def _comparison_arms(report: dict) -> list[dict]:
comparison = report.get("production_cadence_comparison") or {}
arms = list(comparison.get("arms") or [])
if len(arms) != 2:
cadence = (report.get("params") or {}).get("entry_cadence", "unknown")
raise RuntimeError(
f"Expected two live comparison arms for {cadence}; found {len(arms)}"
)
return arms
def _print_arm(row: dict) -> None:
print(
f" {row['arm']}: Sharpe {row.get('sharpe')}, "
f"CAGR {row.get('cagr_pct')}%, DD {row.get('max_drawdown_pct')}%, "
f"trades {row.get('trades')}, skipped cooldown {row.get('skipped_cooldown', 0)}"
)
async def _main() -> None:
args = _parse_args()
snapshot = Path(args.snapshot)
if not snapshot.exists():
raise SystemExit(f"Snapshot not found: {snapshot}")
os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
if args.allow_spawn:
os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
from app.config import settings
from app.services.backtest_service import run_backtest
if args.workers is not None:
settings.backtest_workers = args.workers
out_dir = Path(args.out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
prefix = args.prefix or f"backtest-cadence-{datetime.now():%Y%m%d-%H%M%S}"
engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
reports: dict[str, dict] = {}
try:
async with Session() as db:
for cadence in ("weekly", "daily"):
last_progress: tuple[int, int] | None = None
def progress(done: int, total: int, symbol: str) -> None:
nonlocal last_progress
if args.quiet or last_progress == (done, total):
return
last_progress = (done, total)
label = f" {symbol}" if symbol else ""
print(
f"{cadence} progress: {done}/{total}{label}",
end="\r",
)
reports[cadence] = await run_backtest(
db,
progress_cb=progress,
target_model="production_gtl",
cadence=cadence,
)
if not args.quiet:
print("")
_write_json(out_dir / f"{prefix}-{cadence}.json", reports[cadence])
finally:
await engine.dispose()
arms = [
*_comparison_arms(reports["weekly"]),
*_comparison_arms(reports["daily"]),
]
expected = {
"prod_live_setup_weekly",
"prod_live_setup_daily",
"cooldown_5_weekly",
"cooldown_5_daily",
}
if {row.get("arm") for row in arms} != expected:
raise RuntimeError("The generated cadence report does not contain all four arms")
arm_order = {
"prod_live_setup_weekly": 0,
"prod_live_setup_daily": 1,
"cooldown_5_weekly": 2,
"cooldown_5_daily": 3,
}
arms.sort(key=lambda row: arm_order[str(row["arm"])])
comparison = {
"generated_at": datetime.now().astimezone().isoformat(),
"snapshot": str(snapshot.resolve()),
"target_model": "production_gtl",
"arms": arms,
"full_reports": {
cadence: str((out_dir / f"{prefix}-{cadence}.json").resolve())
for cadence in ("weekly", "daily")
},
"note": (
"All four arms use the same snapshot, activation settings, target model, "
"live Admin exit policy, fees, sizing, and portfolio constraints. Within "
"each cadence pair, only the five-session post-stop lockdown differs."
),
}
comparison_path = out_dir / f"{prefix}-comparison.json"
_write_json(comparison_path, comparison)
print(f"Comparison written: {comparison_path}")
for row in arms:
_print_arm(row)
if __name__ == "__main__":
asyncio.run(_main())
+18 -4
View File
@@ -32,7 +32,10 @@ def _parse_args() -> argparse.Namespace:
parser.add_argument(
"--out",
default=None,
help="JSON report path. Defaults to reports/backtest-<timestamp>.json.",
help=(
"JSON report path. Defaults to "
"reports/backtest-<cadence>-<timestamp>.json."
),
)
parser.add_argument(
"--workers",
@@ -55,6 +58,15 @@ def _parse_args() -> argparse.Namespace:
"structural_sr is a comparison-only chart-S/R model."
),
)
parser.add_argument(
"--cadence",
choices=("weekly", "daily"),
default="weekly",
help=(
"Entry replay cadence. Weekly is the resource-safe production default; "
"daily performs roughly five times as many setup evaluations."
),
)
parser.add_argument(
"--holdout-split",
default=None,
@@ -63,9 +75,9 @@ def _parse_args() -> argparse.Namespace:
return parser.parse_args()
def _default_output_path() -> Path:
def _default_output_path(cadence: str) -> Path:
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
return Path("reports") / f"backtest-{stamp}.json"
return Path("reports") / f"backtest-{cadence}-{stamp}.json"
def _pct(value: Any) -> str:
@@ -93,6 +105,7 @@ def _print_summary(report: dict) -> None:
print("")
print("Backtest summary")
print(f" entry cadence: {(report.get('params') or {}).get('entry_cadence', 'weekly')}")
print(f" candidates: {report.get('candidates')}")
print(f" qualified: {report.get('qualified')}")
print(f" all setups net avg R: {_r(all_setups.get('net_avg_r'))}")
@@ -183,7 +196,7 @@ async def _main() -> None:
if args.workers is not None:
settings.backtest_workers = args.workers
output = Path(args.out) if args.out else _default_output_path()
output = Path(args.out) if args.out else _default_output_path(args.cadence)
output.parent.mkdir(parents=True, exist_ok=True)
engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
@@ -208,6 +221,7 @@ async def _main() -> None:
db,
progress_cb=progress,
target_model=args.target_model,
cadence=args.cadence,
)
finally:
await engine.dispose()
+422
View File
@@ -0,0 +1,422 @@
"""Targeted offline study of a gate-conditioned initial-stop refresh.
The study replays production entries only for the requested period. Whenever
an initial stop is touched, it rebuilds that ticker's setup using bars through
the previous close and recomputes the production momentum gate across the whole
historical universe. If the gate still passes and the new setup has a lower
valid stop, the simulator adopts it and checks it against the same day's low.
This is causal: no value from the stop day's eventual close is used to cancel
an intraday stop. The snapshot is read-only and no live settings are changed.
"""
from __future__ import annotations
import argparse
import asyncio
import bisect
import json
import multiprocessing
import os
import sys
from collections import Counter
from concurrent.futures import ProcessPoolExecutor, as_completed
from datetime import date, datetime
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
def _sqlite_url(path: Path) -> str:
return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
def _parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("snapshot")
parser.add_argument("--start-date", default="2024-07-01")
parser.add_argument("--workers", type=int, default=6)
parser.add_argument("--out", default=None)
parser.add_argument("--quiet", action="store_true")
return parser.parse_args()
def _default_output_path() -> Path:
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
return Path("reports") / f"gate-protected-stop-{stamp}.json"
class GateStopRefresher:
"""Point-in-time gate and replacement-stop calculator for stop events."""
def __init__(
self,
prices: dict[str, tuple],
recommendation_config: dict,
activation: dict,
benchmark_closes: dict[date, float],
) -> None:
from app.services import backtest_service as bt
self.bt = bt
self.prices = prices
self.recommendation_config = recommendation_config
self.activation = activation
self.benchmark_closes = benchmark_closes
self.threshold = float(activation.get("min_momentum_percentile", 80.0))
self.dates = {
symbol: [date.fromordinal(value) for value in columns[0]]
for symbol, columns in prices.items()
}
self.index_of = {
symbol: {value: index for index, value in enumerate(columns[0])}
for symbol, columns in prices.items()
}
self.percentile_cache: dict[int, dict[str, float]] = {}
self.setup_cache: dict[tuple[str, int], dict | None] = {}
self.events: list[dict[str, Any]] = []
def _momentum_percentiles(self, asof_ord: int) -> dict[str, float]:
cached = self.percentile_cache.get(asof_ord)
if cached is not None:
return cached
values: dict[str, float] = {}
for symbol, columns in self.prices.items():
idx = bisect.bisect_right(columns[0], asof_ord) - 1
if idx < 252:
continue
closes = columns[4]
value = self.bt._residual_momentum_12_1(
self.dates[symbol], closes, idx, self.benchmark_closes
)
if value is None and closes[idx - 252] > 0:
value = closes[idx - 21] / closes[idx - 252] - 1.0
if value is not None:
values[symbol] = float(value)
ordered = sorted(values, key=lambda symbol: values[symbol])
denominator = len(ordered) - 1
percentiles = {
symbol: (rank / denominator * 100.0) if denominator > 0 else 100.0
for rank, symbol in enumerate(ordered)
}
self.percentile_cache[asof_ord] = percentiles
return percentiles
def _long_setup(self, symbol: str, asof_idx: int) -> dict | None:
columns = self.prices[symbol]
asof_ord = columns[0][asof_idx]
key = (symbol, asof_ord)
if key in self.setup_cache:
return self.setup_cache[key]
records = [
SimpleNamespace(
date=date.fromordinal(o),
open=op,
high=high,
low=low,
close=close,
volume=volume,
)
for o, op, high, low, close, volume in zip(
columns[0][: asof_idx + 1],
columns[1][: asof_idx + 1],
columns[2][: asof_idx + 1],
columns[3][: asof_idx + 1],
columns[4][: asof_idx + 1],
columns[5][: asof_idx + 1],
)
]
setups = self.bt._window_setups(
records, self.recommendation_config, self.activation
)
setup = next((row for row in setups if row["direction"] == "long"), None)
self.setup_cache[key] = setup
return setup
def __call__(
self,
symbol: str,
stop_ord: int,
active_stop: float,
position: dict,
bar: Any,
) -> float | None:
columns = self.prices[symbol]
stop_idx = self.index_of[symbol].get(stop_ord)
if stop_idx is None:
stop_idx = bisect.bisect_left(columns[0], stop_ord)
asof_idx = stop_idx - 1
if asof_idx < self.bt.MIN_LOOKBACK - 1:
return None
asof_ord = columns[0][asof_idx]
setup = self._long_setup(symbol, asof_idx)
momentum_pct = self._momentum_percentiles(asof_ord).get(symbol)
gate_passed = bool(
setup is not None
and self.bt._momentum_qualifies(
{
"meets_core": setup["meets_core"],
"direction": "long",
self.bt.PRODUCTION_PERCENTILE_KEY: momentum_pct,
},
self.threshold,
)
)
new_stop = float(setup["stop"]) if gate_passed and setup is not None else None
lower_stop = bool(new_stop is not None and new_stop < active_stop - 1e-9)
original_risk = float(position["entry"] - position["initial_stop"])
replacement_risk_r = (
(float(position["entry"]) - new_stop) / original_risk
if lower_stop and original_risk > 0 and new_stop is not None
else None
)
self.events.append({
"symbol": symbol,
"stop_date": date.fromordinal(stop_ord).isoformat(),
"gate_asof_date": date.fromordinal(asof_ord).isoformat(),
"momentum_percentile": round(momentum_pct, 2)
if momentum_pct is not None
else None,
"gate_core_passed": bool(setup and setup["meets_core"]),
"gate_passed": gate_passed,
"active_stop": round(active_stop, 4),
"replacement_stop": round(new_stop, 4) if new_stop is not None else None,
"lower_stop": lower_stop,
"same_bar_survives": bool(lower_stop and bar.low > new_stop),
"replacement_risk_r": round(replacement_risk_r, 3)
if replacement_risk_r is not None
else None,
})
return new_stop
def _arm(label: str, sim: dict) -> dict:
trade_details = sim.pop("trade_details", None)
row = {"arm": label, **sim}
if trade_details is not None:
row["trade_details"] = trade_details
return row
def _rescued_trade_summary(trades: list[dict]) -> dict:
rescued = [trade for trade in trades if trade.get("stop_refreshes", 0) > 0]
rs = [float(trade["r"]) for trade in rescued]
return {
"trades": len(rescued),
"wins": sum(value > 0 for value in rs),
"win_rate": round(sum(value > 0 for value in rs) / len(rs) * 100.0, 1)
if rs
else None,
"avg_r": round(sum(rs) / len(rs), 3) if rs else None,
"total_r": round(sum(rs), 2) if rs else None,
"worst_r": round(min(rs), 2) if rs else None,
"best_r": round(max(rs), 2) if rs else None,
"exit_reasons": dict(Counter(trade["reason"] for trade in rescued)),
}
async def _main() -> None:
args = _parse_args()
snapshot = Path(args.snapshot)
if not snapshot.exists():
raise SystemExit(f"Snapshot not found: {snapshot}")
try:
start_date = date.fromisoformat(args.start_date)
except ValueError as exc:
raise SystemExit("--start-date must use YYYY-MM-DD") from exc
os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
from app.models.ticker import Ticker
from app.services import backtest_service as bt
from app.services.admin_service import get_activation_config
from app.services.paper_trade_service import get_exit_policy
from app.services.recommendation_service import get_recommendation_config
engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
try:
async with Session() as db:
recommendation_config = await get_recommendation_config(db)
activation = await get_activation_config(db)
exit_config = await get_exit_policy(db)
benchmark_closes = await bt._load_benchmark_closes_for_backtest(
db, days=None, refresh=False
)
ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol))
symbols = [ticker.symbol for ticker in ticker_result.scalars().all()]
prices: dict[str, tuple] = {}
for index, symbol in enumerate(symbols, 1):
columns = await bt._fetch_columns(db, symbol)
if columns is not None:
prices[symbol] = columns
if not args.quiet and index % 50 == 0:
print(f"loaded prices: {index}/{len(symbols)}", flush=True)
finally:
await engine.dispose()
candidates: list[dict] = []
workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1))
context = multiprocessing.get_context("spawn")
with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
futures = {
pool.submit(
bt._replay_candidates_for_period,
symbol,
columns,
recommendation_config,
activation,
benchmark_closes,
start_date,
): symbol
for symbol, columns in prices.items()
}
for index, future in enumerate(as_completed(futures), 1):
candidates.extend(future.result())
if not args.quiet and index % 25 == 0:
print(f"replayed tickers: {index}/{len(futures)}", flush=True)
bt._assign_momentum_percentiles(candidates)
bt._assign_residual_momentum_percentiles(candidates)
bt._assign_low_volatility_percentiles(candidates)
bt._assign_activation_momentum_percentiles(candidates)
bt._assign_residual_high_vol_blend(candidates)
threshold = float(activation.get("min_momentum_percentile", 80.0))
for candidate in candidates:
candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)
strategy = next(
row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production")
)
entry_config = bt._entry_variant_config(str(strategy["entry_variant"]))
if entry_config is None:
raise RuntimeError("Production entry configuration missing")
exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get(
str(exit_config.get("mode", "atr_trailing")), "atr_trail3"
)
hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS)))
trail_multiplier = float(
exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)
)
sim_kwargs = {
"qualified_fn": None,
"ranking_key": str(
entry_config.get("ranking_key") or entry_config["percentile_key"]
),
"max_positions": int(entry_config["max_positions"]),
"risk_per_trade": float(entry_config["risk_per_trade"]),
"atr_trail_multiplier": trail_multiplier,
"start_date": start_date,
}
baseline = bt._simulate_portfolio(
candidates, prices, benchmark_closes, exit_policy, hold_days, **sim_kwargs
)
cooldown_5 = bt._simulate_portfolio(
candidates,
prices,
benchmark_closes,
exit_policy,
hold_days,
reentry_cooldown_sessions=5,
**sim_kwargs,
)
cooldown_10 = bt._simulate_portfolio(
candidates,
prices,
benchmark_closes,
exit_policy,
hold_days,
reentry_cooldown_sessions=10,
**sim_kwargs,
)
refresher = GateStopRefresher(
prices, recommendation_config, activation, benchmark_closes
)
gate_protected = bt._simulate_portfolio(
candidates,
prices,
benchmark_closes,
exit_policy,
hold_days,
initial_stop_refresh_fn=refresher,
include_trades=True,
**sim_kwargs,
)
if any(row is None for row in (baseline, cooldown_5, cooldown_10, gate_protected)):
raise RuntimeError("A study arm produced no trades")
gate_trades = list(gate_protected.get("trade_details") or [])
event_counts = Counter()
for event in refresher.events:
event_counts["stop_touches"] += 1
if event["gate_passed"]:
event_counts["gate_passed"] += 1
if event["lower_stop"]:
event_counts["lower_stop"] += 1
if event["same_bar_survives"]:
event_counts["same_bar_survives"] += 1
report = {
"generated_at": datetime.now().astimezone().isoformat(),
"snapshot": str(snapshot.resolve()),
"period_start": start_date.isoformat(),
"tickers": len(prices),
"entry_candidates": len(candidates),
"qualified_candidates": sum(bool(row["qualified"]) for row in candidates),
"params": {
"entry_cadence_days": bt.STEP_DAYS,
"setup_stop_atr_multiplier": bt.ATR_MULTIPLIER,
"exit_policy": exit_policy,
"exit_atr_multiplier": trail_multiplier,
"hold_days": hold_days,
"momentum_percentile_floor": threshold,
"gate_refresh_information_cutoff": "previous close",
},
"arms": [
_arm("baseline", baseline),
_arm("cooldown_5", cooldown_5),
_arm("cooldown_10", cooldown_10),
_arm("gate_protected_stop", gate_protected),
],
"gate_stop_events": {
**dict(event_counts),
"unique_symbols": len({event["symbol"] for event in refresher.events}),
"rescued_trade_outcomes": _rescued_trade_summary(gate_trades),
"events": refresher.events,
},
"note": (
"The gate-protected arm recalculates the gate at an initial-stop touch "
"using only data available through the previous close. It accepts only "
"a lower stop from a newly valid long setup and checks that replacement "
"against the same bar. It does not cancel stops using the later same-day close."
),
}
output = Path(args.out) if args.out else _default_output_path()
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
print(f"Report written: {output}")
for arm in report["arms"]:
print(
f"{arm['arm']}: Sharpe {arm['sharpe']}, CAGR {arm['cagr_pct']}%, "
f"DD {arm['max_drawdown_pct']}%, trades {arm['trades']}"
)
print(f"gate stop events: {dict(event_counts)}")
print(f"rescued outcomes: {report['gate_stop_events']['rescued_trade_outcomes']}")
if __name__ == "__main__":
asyncio.run(_main())
+541
View File
@@ -0,0 +1,541 @@
"""Offline event study for stateful post-stop re-entry policies.
Initial entries keep the validated weekly production cadence. After an initial
stop, the affected ticker is evaluated on every subsequent daily close. This
isolates the exact churn problem without changing the rest of the portfolio.
All arms retain the hard stop, production position sizing, 3x ATR trail, and
round-trip transaction costs.
"""
from __future__ import annotations
import argparse
import asyncio
import bisect
import json
import multiprocessing
import os
import pickle
import sys
from collections import Counter
from concurrent.futures import ProcessPoolExecutor, as_completed
from datetime import date, datetime
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
RECLAIM_ATR_BUFFER = 0.25
def _sqlite_url(path: Path) -> str:
return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
def _parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("snapshot")
parser.add_argument("--start-date", default="2024-07-01")
parser.add_argument("--workers", type=int, default=6)
parser.add_argument("--out", default=None)
parser.add_argument(
"--candidate-cache",
default=None,
help="Optional pickle cache for the expensive weekly candidate replay.",
)
parser.add_argument("--quiet", action="store_true")
parser.add_argument(
"--cooldowns",
type=int,
nargs="+",
default=None,
help=(
"Run an immediate baseline plus the given cooldown lengths instead "
"of the gate-reset policy study (for example: 3 5 7 10)."
),
)
return parser.parse_args()
def _default_output_path() -> Path:
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
return Path("reports") / f"post-stop-reentry-{stamp}.json"
class DailySetupEngine:
"""Point-in-time daily setup and universe-rank cache."""
def __init__(
self,
prices: dict[str, tuple],
recommendation_config: dict,
activation: dict,
benchmark_closes: dict[date, float],
) -> None:
from app.services import backtest_service as bt
self.bt = bt
self.prices = prices
self.recommendation_config = recommendation_config
self.activation = activation
self.benchmark_closes = benchmark_closes
self.threshold = float(activation.get("min_momentum_percentile", 80.0))
self.dates = {
symbol: [date.fromordinal(value) for value in columns[0]]
for symbol, columns in prices.items()
}
self.index_of = {
symbol: {value: index for index, value in enumerate(columns[0])}
for symbol, columns in prices.items()
}
self.rank_cache: dict[int, dict[str, tuple[float, float]]] = {}
self.candidate_cache: dict[tuple[str, int], dict | None] = {}
self.atr_cache: dict[tuple[str, int], float | None] = {}
@staticmethod
def _percentiles(values: dict[str, float]) -> dict[str, float]:
ordered = sorted(values, key=lambda symbol: values[symbol])
denominator = len(ordered) - 1
return {
symbol: (rank / denominator * 100.0) if denominator > 0 else 100.0
for rank, symbol in enumerate(ordered)
}
def _ranks(self, asof_ord: int) -> dict[str, tuple[float, float]]:
cached = self.rank_cache.get(asof_ord)
if cached is not None:
return cached
momentum_values: dict[str, float] = {}
volatility_values: dict[str, float] = {}
for symbol, columns in self.prices.items():
idx = bisect.bisect_right(columns[0], asof_ord) - 1
if idx < 0:
continue
closes = columns[4]
if idx >= 252:
momentum = self.bt._residual_momentum_12_1(
self.dates[symbol], closes, idx, self.benchmark_closes
)
if momentum is None and closes[idx - 252] > 0:
momentum = closes[idx - 21] / closes[idx - 252] - 1.0
if momentum is not None:
momentum_values[symbol] = float(momentum)
volatility = self.bt._realized_vol_6m(closes, idx)
if volatility is not None:
volatility_values[symbol] = float(volatility)
momentum_pct = self._percentiles(momentum_values)
volatility_pct = self._percentiles(volatility_values)
ranks = {
symbol: (momentum_pct[symbol], volatility_pct.get(symbol, 0.0))
for symbol in momentum_pct
}
self.rank_cache[asof_ord] = ranks
return ranks
def atr(self, symbol: str, asof_ord: int) -> float | None:
key = (symbol, asof_ord)
if key in self.atr_cache:
return self.atr_cache[key]
columns = self.prices[symbol]
idx = self.index_of[symbol].get(asof_ord)
if idx is None:
idx = bisect.bisect_right(columns[0], asof_ord) - 1
if idx < 0:
self.atr_cache[key] = None
return None
try:
value = self.bt.compute_atr(
columns[2][: idx + 1],
columns[3][: idx + 1],
columns[4][: idx + 1],
)["atr"]
result = float(value) if value and value > 0 else None
except Exception:
result = None
self.atr_cache[key] = result
return result
def candidate(self, symbol: str, asof_ord: int) -> dict | None:
key = (symbol, asof_ord)
if key in self.candidate_cache:
cached = self.candidate_cache[key]
return dict(cached) if cached is not None else None
columns = self.prices[symbol]
idx = self.index_of[symbol].get(asof_ord)
if idx is None or idx < self.bt.MIN_LOOKBACK - 1:
self.candidate_cache[key] = None
return None
records = [
SimpleNamespace(
date=date.fromordinal(o),
open=op,
high=high,
low=low,
close=close,
volume=volume,
)
for o, op, high, low, close, volume in zip(
columns[0][: idx + 1],
columns[1][: idx + 1],
columns[2][: idx + 1],
columns[3][: idx + 1],
columns[4][: idx + 1],
columns[5][: idx + 1],
)
]
setups = self.bt._window_setups(
records, self.recommendation_config, self.activation
)
setup = next((row for row in setups if row["direction"] == "long"), None)
rank = self._ranks(asof_ord).get(symbol)
gate_passed = bool(
setup is not None
and rank is not None
and self.bt._momentum_qualifies(
{
"meets_core": setup["meets_core"],
"direction": "long",
self.bt.PRODUCTION_PERCENTILE_KEY: rank[0],
},
self.threshold,
)
)
if not gate_passed or setup is None or rank is None:
self.candidate_cache[key] = None
return None
strategy_rank = (
rank[0] * self.bt.STRATEGY_RANK_MOMENTUM_WEIGHT
+ rank[1] * (1.0 - self.bt.STRATEGY_RANK_MOMENTUM_WEIGHT)
)
candidate = {
"symbol": symbol,
"date": date.fromordinal(asof_ord).isoformat(),
"direction": "long",
"entry": float(setup["entry"]),
"stop": float(setup["stop"]),
"target": float(setup["target"]),
"qualified": True,
self.bt.PRODUCTION_PERCENTILE_KEY: rank[0],
self.bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: strategy_rank,
}
self.candidate_cache[key] = candidate
return dict(candidate)
class ReentryPolicy:
def __init__(self, name: str, engine: DailySetupEngine) -> None:
self.name = name
self.engine = engine
self.checks = 0
self.gate_passes = 0
self.emitted = Counter()
def __call__(
self,
symbol: str,
asof_ord: int,
state: dict,
bar: Any,
) -> dict | None:
self.checks += 1
if "reentry_trigger" not in state:
stop_atr = self.engine.atr(symbol, state["stop_ord"])
state["reentry_trigger"] = (
state["stop_day_high"] + RECLAIM_ATR_BUFFER * stop_atr
if stop_atr is not None
else state["stop_day_high"]
)
candidate = self.engine.candidate(symbol, asof_ord)
if candidate is None:
state["gate_went_unqualified"] = True
return None
self.gate_passes += 1
reason: str | None = None
sessions = int(state["sessions_since_stop"])
if self.name == "immediate":
reason = "gate_still_or_again_qualified"
elif self.name.startswith("cooldown_"):
cooldown_sessions = int(self.name.removeprefix("cooldown_"))
if sessions >= cooldown_sessions:
reason = f"{cooldown_sessions}_session_cooldown_complete"
elif self.name == "gate_reset":
if state["gate_went_unqualified"]:
reason = "gate_failed_then_requalified"
elif self.name == "gate_reset_or_reclaim":
if state["gate_went_unqualified"]:
reason = "gate_failed_then_requalified"
elif (
bar.close > state["reentry_trigger"]
and float(candidate["stop"]) > state["previous_stop"]
):
reason = "price_reclaim_with_improved_stop"
else:
raise ValueError(f"Unknown re-entry policy: {self.name}")
if reason is None:
return None
emitted = dict(candidate)
emitted["_reentry_reason"] = reason
self.emitted[reason] += 1
return emitted
def summary(self) -> dict:
return {
"daily_checks": self.checks,
"qualified_checks": self.gate_passes,
"emitted_by_reason": dict(self.emitted),
}
def _trade_summary(trades: list[dict]) -> dict:
reentries = [trade for trade in trades if trade.get("is_reentry")]
waits = [
int(trade["reentry_wait_sessions"])
for trade in reentries
if trade.get("reentry_wait_sessions") is not None
]
return {
"transaction_cost": round(
sum(float(trade["transaction_cost"]) for trade in trades), 2
),
"reentry_trades": len(reentries),
"same_day_reentries": sum(wait == 0 for wait in waits),
"next_day_reentries": sum(wait == 1 for wait in waits),
"reentries_within_5_sessions": sum(wait <= 5 for wait in waits),
"avg_reentry_wait_sessions": round(sum(waits) / len(waits), 1)
if waits
else None,
"reentry_win_rate": round(
sum(float(trade["pnl"]) > 0 for trade in reentries)
/ len(reentries)
* 100.0,
1,
)
if reentries
else None,
"reentry_total_pnl": round(
sum(float(trade["pnl"]) for trade in reentries), 2
),
}
async def _main() -> None:
args = _parse_args()
snapshot = Path(args.snapshot)
if not snapshot.exists():
raise SystemExit(f"Snapshot not found: {snapshot}")
try:
start_date = date.fromisoformat(args.start_date)
except ValueError as exc:
raise SystemExit("--start-date must use YYYY-MM-DD") from exc
os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
from app.models.ticker import Ticker
from app.services import backtest_service as bt
from app.services.admin_service import get_activation_config
from app.services.paper_trade_service import get_exit_policy
from app.services.recommendation_service import get_recommendation_config
engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
try:
async with Session() as db:
recommendation_config = await get_recommendation_config(db)
activation = await get_activation_config(db)
exit_config = await get_exit_policy(db)
benchmark_closes = await bt._load_benchmark_closes_for_backtest(
db, days=None, refresh=False
)
ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol))
symbols = [ticker.symbol for ticker in ticker_result.scalars().all()]
prices: dict[str, tuple] = {}
for index, symbol in enumerate(symbols, 1):
columns = await bt._fetch_columns(db, symbol)
if columns is not None:
prices[symbol] = columns
if not args.quiet and index % 50 == 0:
print(f"loaded prices: {index}/{len(symbols)}", flush=True)
finally:
await engine.dispose()
cache_path = Path(args.candidate_cache) if args.candidate_cache else None
snapshot_stat = snapshot.stat()
cache_key = {
"snapshot": str(snapshot.resolve()),
"snapshot_size": snapshot_stat.st_size,
"snapshot_mtime_ns": snapshot_stat.st_mtime_ns,
"start_date": start_date.isoformat(),
}
candidates: list[dict]
if cache_path is not None and cache_path.exists():
with cache_path.open("rb") as handle:
cached_replay = pickle.load(handle) # noqa: S301 - trusted local cache
if cached_replay.get("key") != cache_key:
raise SystemExit(f"Candidate cache does not match this run: {cache_path}")
candidates = list(cached_replay["candidates"])
if not args.quiet:
print(f"loaded candidate cache: {cache_path}", flush=True)
else:
candidates = []
workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1))
context = multiprocessing.get_context("spawn")
with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
futures = {
pool.submit(
bt._replay_candidates_for_period,
symbol,
columns,
recommendation_config,
activation,
benchmark_closes,
start_date,
): symbol
for symbol, columns in prices.items()
}
for index, future in enumerate(as_completed(futures), 1):
candidates.extend(future.result())
if not args.quiet and index % 25 == 0:
print(f"replayed tickers: {index}/{len(futures)}", flush=True)
if cache_path is not None:
cache_path.parent.mkdir(parents=True, exist_ok=True)
with cache_path.open("wb") as handle:
pickle.dump(
{"key": cache_key, "candidates": candidates},
handle,
protocol=pickle.HIGHEST_PROTOCOL,
)
if not args.quiet:
print(f"wrote candidate cache: {cache_path}", flush=True)
bt._assign_momentum_percentiles(candidates)
bt._assign_residual_momentum_percentiles(candidates)
bt._assign_low_volatility_percentiles(candidates)
bt._assign_activation_momentum_percentiles(candidates)
bt._assign_residual_high_vol_blend(candidates)
threshold = float(activation.get("min_momentum_percentile", 80.0))
for candidate in candidates:
candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)
strategy = next(
row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production")
)
entry_config = bt._entry_variant_config(str(strategy["entry_variant"]))
if entry_config is None:
raise RuntimeError("Production entry configuration missing")
exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get(
str(exit_config.get("mode", "atr_trailing")), "atr_trail3"
)
hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS)))
trail_multiplier = float(
exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)
)
sim_kwargs = {
"ranking_key": str(
entry_config.get("ranking_key") or entry_config["percentile_key"]
),
"max_positions": int(entry_config["max_positions"]),
"risk_per_trade": float(entry_config["risk_per_trade"]),
"atr_trail_multiplier": trail_multiplier,
"start_date": start_date,
"include_trades": True,
}
daily_engine = DailySetupEngine(
prices, recommendation_config, activation, benchmark_closes
)
if args.cooldowns is None:
policy_names = (
"immediate",
"cooldown_5",
"gate_reset",
"gate_reset_or_reclaim",
)
else:
cooldowns = sorted(set(args.cooldowns))
if any(value < 1 for value in cooldowns):
raise SystemExit("--cooldowns values must be positive integers")
policy_names = ("immediate", *(f"cooldown_{value}" for value in cooldowns))
arms: list[dict] = []
for policy_name in policy_names:
policy = ReentryPolicy(policy_name, daily_engine)
sim = bt._simulate_portfolio(
candidates,
prices,
benchmark_closes,
exit_policy,
hold_days,
post_stop_reentry_fn=policy,
**sim_kwargs,
)
if sim is None:
raise RuntimeError(f"Policy {policy_name} produced no trades")
trades = list(sim.pop("trade_details"))
arms.append({
"arm": policy_name,
**sim,
"turnover": _trade_summary(trades),
"policy": policy.summary(),
"trade_details": trades,
})
output = Path(args.out) if args.out else _default_output_path()
report = {
"generated_at": datetime.now().astimezone().isoformat(),
"snapshot": str(snapshot.resolve()),
"period_start": start_date.isoformat(),
"tickers": len(prices),
"entry_candidates": len(candidates),
"qualified_candidates": sum(bool(row["qualified"]) for row in candidates),
"params": {
"initial_entry_cadence_days": bt.STEP_DAYS,
"post_stop_evaluation_cadence_days": 1,
"setup_stop_atr_multiplier": bt.ATR_MULTIPLIER,
"exit_policy": exit_policy,
"exit_atr_multiplier": trail_multiplier,
"hold_days": hold_days,
"cost_per_side_pct": bt.COST_PER_SIDE * 100.0,
"momentum_percentile_floor": threshold,
"reclaim_atr_buffer": RECLAIM_ATR_BUFFER,
"cooldown_sessions": cooldowns if args.cooldowns is not None else None,
},
"arms": arms,
"note": (
"Initial opportunities retain the validated weekly replay cadence. "
"Only tickers stopped at their initial stop switch to daily evaluation, "
"which isolates next-day/same-episode re-entry churn. A cooldown of N "
"sessions permits the first re-entry at wait_sessions=N. Gate reset "
"requires at least one unqualified daily close before requalification. "
"The reclaim arm alternatively accepts a close above stop-day high + "
"0.25 ATR only when the new setup stop is above the prior stop."
),
}
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
print(f"Report written: {output}")
for arm in arms:
turnover = arm["turnover"]
print(
f"{arm['arm']}: Sharpe {arm['sharpe']}, CAGR {arm['cagr_pct']}%, "
f"DD {arm['max_drawdown_pct']}%, trades {arm['trades']}, "
f"reentries {turnover['reentry_trades']}, fees ${turnover['transaction_cost']}"
)
if __name__ == "__main__":
asyncio.run(_main())
+283
View File
@@ -575,6 +575,228 @@ class TestSimulatePortfolio:
assert sim["trades"] == 1
assert sim["worst_trade_r"] == pytest.approx(-2.0) # (90 100) / 5
def test_initial_stop_cooldown_blocks_immediate_reentry(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
candidates = [
_sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
_sim_cand("AAA", self.ORD + 1, entry=94.0, stop=89.0, target=110.0),
]
baseline = bt._simulate_portfolio(candidates, prices, None, "hold", 30)
cooldown = bt._simulate_portfolio(
candidates,
prices,
None,
"hold",
30,
reentry_cooldown_sessions=5,
)
assert baseline is not None and baseline["trades"] == 2
assert cooldown is not None and cooldown["trades"] == 1
assert cooldown["skipped_cooldown"] == 1
assert cooldown["reentry_cooldown_sessions"] == 5
def test_initial_stop_cooldown_unlocks_exactly_after_session_five(self):
closes = [100.0, 94.0, 96.0, 96.0, 96.0, 96.0, 97.0, 98.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
candidates = [
_sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
# Four completed sessions since the stop: still locked.
_sim_cand("AAA", self.ORD + 5, entry=96.0, stop=90.0, target=115.0),
# Five completed sessions since the stop: first permitted re-entry.
_sim_cand("AAA", self.ORD + 6, entry=97.0, stop=90.0, target=118.0),
]
sim = bt._simulate_portfolio(
candidates,
prices,
None,
"hold",
30,
reentry_cooldown_sessions=5,
include_trades=True,
)
assert sim is not None
assert sim["trades"] == 2
assert sim["skipped_cooldown"] == 1
assert sim["trade_details"][1]["entry_date"] == date.fromordinal(
self.ORD + 6
).isoformat()
def test_production_monitor_applies_live_reentry_lockdown(self, monkeypatch):
def fake_simulator(*_args, **kwargs):
return {
"trades": 0,
"applied_reentry_lockdown": kwargs.get("reentry_cooldown_sessions", 0),
}
monkeypatch.setattr(bt, "_simulate_portfolio", fake_simulator)
market_ord = date(2026, 7, 1).toordinal()
prices = {"AAA": ([market_ord], [], [], [], [], [])}
monitor = bt._portfolio_monitor([], prices, None, 30)
production_rows = [
row for row in monitor["runs"] if row["is_production"]
]
comparison_rows = [
row for row in monitor["runs"] if not row["is_production"]
]
assert production_rows
assert all(
row["reentry_lockdown_sessions"] == bt.REENTRY_LOCKDOWN_SESSIONS
and row["applied_reentry_lockdown"] == bt.REENTRY_LOCKDOWN_SESSIONS
for row in production_rows
)
assert comparison_rows
assert all(
row["reentry_lockdown_sessions"] == 0
and row["applied_reentry_lockdown"] == 0
for row in comparison_rows
)
def test_production_cadence_comparison_names_exact_two_arms(self):
monitor = {
"runs": [
{
"comparison_arm": "live_no_lockdown",
"lookback": "all",
"reentry_lockdown_sessions": 0,
"trades": 10,
"equity_curve": [{"date": "2026-01-01", "value": 1.0}],
},
{
"comparison_arm": "live_lockdown_5",
"lookback": "all",
"reentry_lockdown_sessions": 5,
"trades": 8,
"benchmark_curve": [{"date": "2026-01-01", "value": 1.0}],
},
]
}
comparison = bt._production_cadence_comparison(monitor, "daily")
assert comparison is not None
assert [row["arm"] for row in comparison["arms"]] == [
"prod_live_setup_daily",
"cooldown_5_daily",
]
assert all("equity_curve" not in row for row in comparison["arms"])
assert all("benchmark_curve" not in row for row in comparison["arms"])
def test_initial_stop_can_refresh_lower_and_survive_same_bar(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
candidate = _sim_cand(
"AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
)
sim = bt._simulate_portfolio(
[candidate],
prices,
None,
"hold",
2,
initial_stop_refresh_fn=lambda *_: 90.0,
include_trades=True,
)
assert sim is not None
assert sim["stop_refresh_attempts"] == 1
assert sim["stop_refreshes"] == 1
assert sim["stop_refresh_same_bar_hits"] == 0
assert sim["exit_reasons"] == {"time": 1}
assert sim["trade_details"][0]["stop_refreshes"] == 1
def test_refreshed_stop_is_checked_against_same_bar(self):
ords = list(range(self.ORD, self.ORD + 2))
prices = {
"AAA": (
ords,
[100.0, 94.0],
[101.0, 96.0],
[99.0, 89.0],
[100.0, 94.0],
[1, 1],
)
}
candidate = _sim_cand(
"AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
)
sim = bt._simulate_portfolio(
[candidate],
prices,
None,
"hold",
30,
initial_stop_refresh_fn=lambda *_: 90.0,
)
assert sim is not None
assert sim["stop_refresh_same_bar_hits"] == 1
assert sim["worst_trade_r"] == pytest.approx(-2.0)
def test_post_stop_state_suppresses_same_episode_candidate(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
candidates = [
_sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
_sim_cand("AAA", self.ORD + 1, entry=94.0, stop=89.0, target=110.0),
]
sim = bt._simulate_portfolio(
candidates,
prices,
None,
"hold",
30,
post_stop_reentry_fn=lambda *_: None,
)
assert sim is not None
assert sim["trades"] == 1
assert sim["post_stop_events"] == 1
assert sim["post_stop_reentries"] == 0
assert sim["post_stop_states_open_at_end"] == 1
def test_post_stop_callback_can_reenter_same_day(self):
closes = [100.0, 94.0, 96.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
initial = _sim_cand(
"AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
)
def immediate_reentry(sym, current_ord, _state, bar):
return _sim_cand(
sym,
current_ord,
entry=bar.close,
stop=bar.close - 5.0,
target=bar.close + 15.0,
)
sim = bt._simulate_portfolio(
[initial],
prices,
None,
"hold",
30,
post_stop_reentry_fn=immediate_reentry,
include_trades=True,
)
assert sim is not None
assert sim["trades"] == 2
assert sim["post_stop_reentries"] == 1
assert sim["reentry_events"][0]["wait_sessions"] == 0
assert sim["trade_details"][1]["is_reentry"] is True
assert sim["trade_details"][1]["reentry_wait_sessions"] == 0
def test_sma50_policy_exits_on_close_break(self):
closes = [100.0] * 56 + [90.0, 91.0]
prices = {"AAA": _sim_prices(self.ORD, closes)}
@@ -722,6 +944,7 @@ def test_build_recommendation_prefers_production_monitor_headline():
})
assert rec["headline"] is not None
assert "3x ATR trailing exit" in rec["headline"]
assert "5-session re-entry lockdown" in rec["headline"]
assert any(item["topic"] == "production" for item in rec["items"])
@@ -736,6 +959,15 @@ def test_backtest_target_model_is_small_and_validated():
bt.validate_backtest_target_model("legacy_range_grid_touch")
def test_backtest_cadence_is_small_validated_and_session_based():
assert bt.validate_backtest_cadence(" WEEKLY ") == "weekly"
assert bt.validate_backtest_cadence("daily") == "daily"
assert bt.backtest_step_sessions("weekly") == 5
assert bt.backtest_step_sessions("daily") == 1
with pytest.raises(ValueError, match="Unknown backtest cadence"):
bt.validate_backtest_cadence("monthly")
def _flat_window_records():
return [
SimpleNamespace(
@@ -829,6 +1061,46 @@ def test_replay_ticker_candidates_carry_gate_fields():
assert c.get("action") is not None
assert "risk_level" in c
assert c["target_model"] == bt.PRODUCTION_GTL_TARGET_MODEL
assert c["ranking_period"][0] == "week"
daily_cands = bt._replay_ticker(
"OSC",
bars,
dict(DEFAULT_RECOMMENDATION_CONFIG),
dict(ACTIVATION_DEFAULTS),
cadence="daily",
)
assert len(daily_cands) > len(cands)
assert all(c["ranking_period"][0] == "date" for c in daily_cands)
def test_daily_replay_uses_exact_date_ranking_periods():
candidates = [
{
"iso_week": (2026, 1),
"ranking_period": ("date", date(2026, 1, 5).toordinal()),
"momentum": 0.10,
},
{
"iso_week": (2026, 1),
"ranking_period": ("date", date(2026, 1, 5).toordinal()),
"momentum": 0.20,
},
{
"iso_week": (2026, 1),
"ranking_period": ("date", date(2026, 1, 6).toordinal()),
"momentum": 0.90,
},
{
"iso_week": (2026, 1),
"ranking_period": ("date", date(2026, 1, 6).toordinal()),
"momentum": 0.30,
},
]
bt._assign_momentum_percentiles(candidates)
assert [row["momentum_percentile"] for row in candidates] == [0.0, 100.0, 100.0, 0.0]
async def _seed_oscillating_ticker(session, symbol: str, n: int = 160) -> None:
@@ -870,12 +1142,23 @@ async def test_run_backtest_smoke(session):
assert report["params"]["cost_per_side_pct"] == pytest.approx(bt.COST_PER_SIDE * 100)
assert report["params"]["target_model"] == bt.PRODUCTION_GTL_TARGET_MODEL
assert report["params"]["is_production_target_model"] is True
assert report["params"]["entry_cadence"] == "weekly"
assert report["params"]["step_sessions"] == 5
assert (
report["params"]["production_reentry_lockdown_sessions"]
== bt.REENTRY_LOCKDOWN_SESSIONS
)
assert "net_avg_r" in report["overall_all"]
# ablation baseline reproduces the qualified set exactly, and every row
# carries the hold-to-horizon grading alongside the target model
ablation = {r["variant"]: r for r in report["gate_ablation"]}
assert ablation["all_floors"]["total"] == report["overall_qualified"]["total"]
daily_report = await bt.run_backtest(session, cadence="daily")
assert daily_report["params"]["entry_cadence"] == "daily"
assert daily_report["params"]["step_sessions"] == 1
assert daily_report["candidates"] > report["candidates"]
for row in report["gate_ablation"]:
assert "hold_net_avg_r" in row
+68
View File
@@ -48,6 +48,74 @@ async def test_create_and_list_open(session):
assert row["current_price"] == 110.0 # marked to the latest close
async def test_create_trade_enforces_post_stop_lockdown_at_service_boundary(session):
blocked_id = await _seed(session, "LOCKQ", close=100.0)
released_id = await _seed(session, "FREEQ", close=100.0)
today = date.today()
market_sessions = [
today - timedelta(days=8),
today - timedelta(days=7),
today - timedelta(days=6),
today - timedelta(days=3),
today - timedelta(days=2),
today - timedelta(days=1),
]
for market_date in market_sessions:
session.add(BenchmarkPrice(symbol="SPY", date=market_date, close=400.0))
def stopped_trade(ticker_id: int, closed_on: date) -> PaperTrade:
return PaperTrade(
user_id=1,
ticker_id=ticker_id,
direction="long",
entry_price=100.0,
shares=10.0,
stop_loss=95.0,
target=115.0,
status="closed",
opened_at=datetime.combine(
closed_on - timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc
),
close_price=95.0,
closed_at=datetime.combine(
closed_on, datetime.min.time(), tzinfo=timezone.utc
),
close_reason="stop",
)
session.add_all(
[
stopped_trade(blocked_id, market_sessions[1]),
stopped_trade(released_id, market_sessions[0]),
]
)
await session.commit()
with pytest.raises(ValidationError, match="1 market session remaining"):
await svc.create_trade(
session,
1,
symbol="LOCKQ",
direction="long",
entry_price=100.0,
shares=10.0,
stop_loss=95.0,
target=115.0,
)
trade = await svc.create_trade(
session,
1,
symbol="FREEQ",
direction="long",
entry_price=100.0,
shares=10.0,
stop_loss=95.0,
target=115.0,
)
assert trade.ticker_id == released_id
async def test_close_uses_current_price(session):
await _seed(session, "AAA", close=112.0)
trade = await svc.create_trade(session, 1, symbol="AAA", direction="long",
+118
View File
@@ -20,6 +20,7 @@ from hypothesis import given, settings, HealthCheck, strategies as st
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.benchmark_price import BenchmarkPrice
from app.models.ohlcv import OHLCVRecord
from app.models.paper_trade import PaperTrade
from app.models.signal_context_snapshot import SignalContextSnapshot
@@ -607,6 +608,123 @@ async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades(
assert [row["symbol"] for row in ticker_rows] == ["OPENQ"]
@pytest.mark.asyncio
async def test_get_trade_setups_applies_five_session_initial_stop_lockdown(
db_session: AsyncSession,
):
now = datetime.now(timezone.utc)
today = now.date()
if await db_session.get(User, 1) is None:
db_session.add(
User(id=1, username="u", password_hash="x", role="user", has_access=True)
)
await db_session.flush()
blocked = Ticker(symbol="STOP4")
released = Ticker(symbol="STOP5")
trailing = Ticker(symbol="TRAILQ")
db_session.add_all([blocked, released, trailing])
await db_session.flush()
# Six SPY sessions D0..D5 form the canonical market calendar. A stop on
# D0 has five later sessions and is released; a stop on D1 has only four.
market_sessions = [
today - timedelta(days=8),
today - timedelta(days=7),
today - timedelta(days=6),
today - timedelta(days=3),
today - timedelta(days=2),
today - timedelta(days=1),
]
for market_date in market_sessions:
db_session.add(
BenchmarkPrice(
symbol="SPY",
date=market_date,
close=400.0,
)
)
# A bar from an unrelated/scanner-specific calendar must not release the
# ticker one session early. The old universe-wide DISTINCT query did.
db_session.add(
OHLCVRecord(
ticker_id=blocked.id,
date=today,
open=100.0,
high=101.0,
low=99.0,
close=100.0,
volume=1_000,
)
)
for ticker in (blocked, released, trailing):
db_session.add(
TradeSetup(
ticker_id=ticker.id,
direction="long",
entry_price=100.0,
stop_loss=95.0,
target=115.0,
rr_ratio=3.0,
composite_score=80.0,
detected_at=now,
)
)
def closed_trade(ticker: Ticker, closed_on: date, reason: str) -> PaperTrade:
return PaperTrade(
user_id=1,
ticker_id=ticker.id,
direction="long",
entry_price=100.0,
shares=10.0,
stop_loss=95.0,
target=115.0,
status="closed",
opened_at=datetime.combine(
closed_on - timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc
),
close_price=95.0,
closed_at=datetime.combine(
closed_on, datetime.min.time(), tzinfo=timezone.utc
),
close_reason=reason,
)
db_session.add_all(
[
closed_trade(blocked, market_sessions[1], "stop"),
closed_trade(released, market_sessions[0], "stop"),
closed_trade(trailing, market_sessions[-1], "trailing"),
]
)
await db_session.flush()
default_symbols = {
row["symbol"] for row in await get_trade_setups(db_session)
}
assert {"STOP4", "STOP5", "TRAILQ"}.issubset(default_symbols)
available_symbols = {
row["symbol"]
for row in await get_trade_setups(
db_session,
exclude_reentry_lockdown_tickers=True,
)
}
assert "STOP4" not in available_symbols
assert {"STOP5", "TRAILQ"}.issubset(available_symbols)
annotated = await get_trade_setups(
db_session,
symbol="STOP4",
include_reentry_lockdown=True,
)
assert len(annotated) == 1
assert annotated[0]["reentry_lockdown_remaining_sessions"] == 1
async def _seed_stale_setup_with_current_scores(db_session: AsyncSession) -> TradeSetup:
"""Stored setup frozen at scan time (conf 82, neutral) vs. current context
(bullish sentiment, composite 96) that yields live confidence 97.
+11
View File
@@ -3,11 +3,13 @@
import pytest
from app.scheduler import (
_consume_backtest_options,
_consume_backtest_target_model,
_parse_frequency,
_resume_tickers,
_last_successful,
configure_scheduler,
queue_backtest_options,
queue_backtest_target_model,
scheduler,
)
@@ -24,6 +26,15 @@ def test_manual_backtest_target_model_rejects_removed_research_arms():
queue_backtest_target_model("production_control")
def test_manual_backtest_options_are_one_shot_and_default_back_to_weekly():
assert queue_backtest_options("structural_sr", "daily") == (
"structural_sr",
"daily",
)
assert _consume_backtest_options() == ("structural_sr", "daily")
assert _consume_backtest_options() == ("production_gtl", "weekly")
class TestParseFrequency:
def test_hourly(self):
assert _parse_frequency("hourly") == {"hours": 1}