harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research: 3-year stake-sizing campaign + violent-move studies (nothing changes signals)

Browse Source 
Walk-forward "second opinion" stake multiplier (meta-labelling on 65 decision-time
features, 3yr Oanda books @ $0.65) FAILED the pre-registered rule on M5 (clean null)
and M15 (lift inside the label-shuffle null, explained by BUY-vs-SELL). Nothing that
changes numbers ships; the 12-month M5 book re-run is byte-identical to the pinned one.

Behaviour-neutral plumbing:
- position_sizing.build_features (65 features; multiplier path returns 1.0, no artifact)
- SignalReport.penalty_tags / penalty_total / momentum_fading (record-keeping)
- TradeSetup.stake_multiplier = None (reserved)
- gpu_backtest --export-features (default export stays the pinned 63 columns; extras
  forced to the tail even when the first trade lacks an indicator)

Research tools (backtest/research): book_report, burst_map, burst_precursors,
open_trade_bursts, calendar_impact, burst_indicator_map, sizing_model, watch_rule_eval.
Findings recorded in backtest/research/sizing_campaign_2026-09.md: M5 stretch measures
are positive in both halves; every open-trade protection against violent bars loses on
both TFs; high-impact USD releases cause 20-31% of violent bars but trades held through
them do better; violent bars come when price is already at an extreme, not from calm.

Module 29 tests (13). Suite: 435 passed.

Claude-Session: https://claude.ai/code/session_01RWV2yLt3YXHHGVYHSQU2EE
This commit is contained in:
Harwinder Sekhon
2026-09-05 21:19:24 -04:00
parent bed8b549ac
commit 6bf58fd9d4
16 changed files with 4076 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
position_sizing.py
+622
View File
@@ -0,0 +1,622 @@
"""
ForexBot2 — Position sizing (stake-multiplier overlay)
======================================================
Single source of truth for **decision-time features** and the resulting
**stake multiplier**. Everything here is designed to be called from
inside the signal path so that live and backtest see byte-identical
inputs (rule 4: live == backtest parity).
Design contract
---------------
* **Bounded lookbacks.** No feature may read further back than the live
fetch depth of its timeframe (``LIVE_DEPTH_BARS``). The primary frame
is hard-capped at ``MAX_LOOKBACK_BARS`` rows, confirmation frames at
60 rows, and the burst scan at 200 rows, so a feature computed on a
full backtest frame equals the same feature computed on the shorter
live tail.
* **Closed bars only.** The frames handed in are exactly the frames
``analyze_from_dataframes`` received: UTC tz-aware, oldest-first,
capitalized OHLCV columns, forming candle already dropped. The last
row is the decision bar; nothing peeks past it.
* **Never raises into a scan.** ``build_features`` and
``stake_multiplier`` swallow every exception (logged at DEBUG) and
fall back to neutral defaults. A sizing failure can never break a
signal.
* **Advisory today.** ``stake_multiplier`` returns ``1.0`` in every case
at this stage — artifact evaluation lands in a later step. The guard
order is already implemented so the wiring is testable now.
Dependencies are deliberately limited to numpy/pandas plus ``config``
(the ``Timeframe`` enum): this module ships live and must not drag the
provider/news stack into any process that only wants features.
VERDICT 2026-09-06 (backtest/research/sizing_campaign_2026-09.md): the walk-forward
stake-multiplier model FAILED the pre-registered acceptance rule on both M5 and
M15 (M5 a clean null; M15 lift inside the label-shuffle null and explained by
BUY-vs-SELL). NO artifact ships and nothing reads a multiplier live. This module
is research plumbing: `build_features` feeds the backtester's --export-features
flag; `stake_multiplier` returns 1.0 unless a validated artifact is ever added
under a NEW pre-registration.
"""
from __future__ import annotations
import json
import logging
import math
from pathlib import Path
from typing import Dict, Iterable, Optional, Sequence, Tuple
import numpy as np
import pandas as pd
from config import SignalStrength, Timeframe
logger = logging.getLogger(__name__)
#: Bump whenever FEATURE_ORDER or any feature's definition changes.
#: A trained artifact carrying a different value is ignored.
FEATURE_VERSION = 1
#: Bars each timeframe actually holds in LIVE mode. Any feature whose
#: lookback exceeds its timeframe's depth here would be unreproducible
#: live, so the windows below stay far inside these numbers.
LIVE_DEPTH_BARS: Dict[Timeframe, int] = {
Timeframe.M5: 5000,
Timeframe.M15: 5000,
Timeframe.H1: 1460,
Timeframe.H4: 380,
Timeframe.D1: 300,
}
#: Hard cap on any primary-timeframe window used in this module.
MAX_LOOKBACK_BARS = 500
#: Bars of each confirmation timeframe used by the cross-TF features.
CONFIRM_LOOKBACK_BARS = 60
#: Window for the "bars since the last volatility burst" feature. Also
#: the sentinel returned when no burst is present in the window.
BURST_LOOKBACK_BARS = 200
#: Burst threshold: bar range >= this multiple of the PRIOR bar's ATR.
BURST_ATR_MULTIPLE = 3.0
#: Candle duration in seconds. Deliberate local copy of
#: ``providers.base.BaseDataProvider.CANDLE_SECONDS`` — importing the
#: providers package here would pull requests/twelvedata into every
#: feature-only consumer. A test asserts the copies agree.
CANDLE_SECONDS: Dict[Timeframe, int] = {
Timeframe.M5: 300,
Timeframe.M15: 900,
Timeframe.H1: 3600,
Timeframe.H4: 14400,
Timeframe.D1: 86400,
}
#: Grade -> ordinal used by ``f_quality_rank`` (D/F both fall to 0).
_QUALITY_RANK = {"A+": 4.0, "A": 3.0, "B": 2.0, "C": 1.0}
_BULLISH = (SignalStrength.BUY, SignalStrength.STRONG_BUY)
_BEARISH = (SignalStrength.SELL, SignalStrength.STRONG_SELL)
FEATURE_ORDER: Tuple[str, ...] = (
# ── Time ──────────────────────────────────────────────────────
"f_hour_utc",
"f_hour_sin",
"f_hour_cos",
"f_weekday",
"f_sess_asian",
"f_sess_london",
"f_sess_overlap",
"f_sess_ny",
"f_minutes_to_hour",
"f_day_of_month",
"f_first_friday",
# ── Volatility / setup geometry ───────────────────────────────
"f_atr",
"f_atr_pct",
"f_atr_rel_500",
"f_range_pctile_500",
"f_bar_range_atr",
"f_close_pos_in_bar",
"f_risk_price",
"f_sl_atr",
"f_tp1_atr",
"f_rr_planned",
# ── Exhaustion block ──────────────────────────────────────────
"f_swing_dist_atr",
"f_ema21_dist_atr",
"f_ema200_dist_atr",
"f_ema200_dist_pct",
"f_exhaustion_count",
"f_tag_exhaustion",
"f_tag_no_pullback",
"f_tag_weak_pullback",
"f_tag_rsi_extended",
"f_tag_bb_overextended",
"f_tag_ema200_near",
"f_tag_ema200_far",
"f_penalty_total",
"f_momentum_fading",
# ── Signal quality ────────────────────────────────────────────
"f_score_dir",
"f_confidence",
"f_quality_rank",
"f_ind_agree_frac",
"f_ind_oppose_frac",
"f_is_buy",
"f_strat_atr",
"f_strat_pb",
"f_strat_cp",
"f_strat_generic",
# ── Multi-timeframe ───────────────────────────────────────────
"f_mtf_primary",
"f_mtf_c1",
"f_mtf_c2",
"f_mtf_agreement",
# ── Primary-TF indicators ─────────────────────────────────────
"f_rsi",
"f_rsi_dir",
"f_bb_pos",
"f_bb_pos_dir",
"f_adx",
"f_kc_ext_atr",
"f_roc_dir",
"f_macd_hist_dir",
# ── Price path ────────────────────────────────────────────────
"f_ret_6_atr",
"f_ret_12_atr",
"f_ret_24_atr",
"f_ret_48_atr",
"f_consec_closes_dir",
"f_bars_since_burst",
# ── Cross-timeframe extension ─────────────────────────────────
"f_h1_ext_atr",
"f_c_other_ext_atr",
)
#: Value each feature falls back to when it cannot be computed. Only
#: ``f_bars_since_burst`` has a non-zero sentinel (its "none in window"
#: cap); everything else degrades to a neutral 0.0.
_DEFAULTS: Dict[str, float] = {name: 0.0 for name in FEATURE_ORDER}
_DEFAULTS["f_bars_since_burst"] = float(BURST_LOOKBACK_BARS)
#: Penalty tags recorded by SignalAggregator._compute_final_signal that
#: this module exposes as one-hot features.
_TAG_FEATURES = (
("f_tag_exhaustion", "exhaustion"),
("f_tag_no_pullback", "no_pullback"),
("f_tag_weak_pullback", "weak_pullback"),
("f_tag_rsi_extended", "rsi_extended"),
("f_tag_bb_overextended", "bb_overextended"),
("f_tag_ema200_near", "ema200_near"),
("f_tag_ema200_far", "ema200_far"),
)
# ══════════════════════════════════════════════════════════════════
# Small helpers
# ══════════════════════════════════════════════════════════════════
def _finite(value, default: float = 0.0) -> float:
"""Coerce to a finite Python float, falling back to ``default``."""
try:
out = float(value)
except (TypeError, ValueError):
return float(default)
if not math.isfinite(out):
return float(default)
return out
def wilder_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
"""Wilder ATR, identical convention to ``engine._compute_atr``.
True Range, then an EWM with ``alpha = 1/period``,
``min_periods = period`` and ``adjust=False``.
"""
high, low, close = df["High"], df["Low"], df["Close"]
tr1 = high - low
tr2 = (high - close.shift()).abs()
tr3 = (low - close.shift()).abs()
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
return tr.ewm(alpha=1.0 / period, min_periods=period, adjust=False).mean()
def decision_time(primary_df: pd.DataFrame, tf: Timeframe) -> pd.Timestamp:
"""The clock every gate uses: last CLOSED bar label + its duration."""
last = pd.Timestamp(primary_df.index[-1])
return last + pd.Timedelta(seconds=CANDLE_SECONDS.get(tf, 3600))
def session_label(hour_utc: int) -> str:
"""Trading-session bucket for a UTC hour."""
h = int(hour_utc)
if 3 <= h <= 6:
return "asian"
if 7 <= h <= 12:
return "london"
if 13 <= h <= 16:
return "overlap"
if 17 <= h <= 22:
return "ny"
return "off"
# ══════════════════════════════════════════════════════════════════
# Artifact plumbing
# ══════════════════════════════════════════════════════════════════
#: path -> (mtime, artifact-or-None)
_ARTIFACT_CACHE: Dict[str, Tuple[float, Optional[dict]]] = {}
def artifact_path_for(primary_tf: Timeframe, base_dir) -> Path:
"""Where the trained artifact for one primary timeframe lives."""
return Path(base_dir) / f"xauusd_{primary_tf.value}.json"
def load_artifact(path) -> Optional[dict]:
"""Load a sizing artifact (JSON object), cached on (path, mtime).
Returns ``None`` when the file is missing, unreadable, not JSON, or
not a JSON object. Never raises.
"""
try:
p = Path(path)
mtime = p.stat().st_mtime
except Exception:
return None
key = str(p)
cached = _ARTIFACT_CACHE.get(key)
if cached is not None and cached[0] == mtime:
return cached[1]
data: Optional[dict]
try:
with p.open("r", encoding="utf-8") as fh:
loaded = json.load(fh)
data = loaded if isinstance(loaded, dict) else None
except Exception:
logger.debug("position_sizing: artifact %s is unreadable", key,
exc_info=True)
data = None
_ARTIFACT_CACHE[key] = (mtime, data)
return data
# ══════════════════════════════════════════════════════════════════
# Feature construction
# ══════════════════════════════════════════════════════════════════
def _direction(setup, report) -> float:
"""+1.0 for a long, -1.0 for a short — from the TRADE, not the score."""
direction = getattr(setup, "direction", None)
if direction is None:
return 1.0 if _finite(getattr(report, "final_score", 0.0)) >= 0 else -1.0
return 1.0 if str(direction).upper() == "BUY" else -1.0
def _ext_vs_sma(df: pd.DataFrame, d: float) -> float:
"""Direction-signed distance from the 20-bar SMA in ATR units.
Computed on the LAST ``CONFIRM_LOOKBACK_BARS`` rows only, so the
value is identical on a full history and on its live-depth tail.
"""
if df is None or len(df) < 21:
return 0.0
tail = df.iloc[-CONFIRM_LOOKBACK_BARS:]
sma = tail["Close"].rolling(20).mean().iloc[-1]
atr = wilder_atr(tail, 14).iloc[-1]
atr_f = _finite(atr, 0.0)
if atr_f <= 0:
return 0.0
return d * (_finite(tail["Close"].iloc[-1]) - _finite(sma)) / atr_f
def _fill_time(out: Dict[str, float], dt: pd.Timestamp) -> None:
hour = int(dt.hour)
out["f_hour_utc"] = float(hour)
out["f_hour_sin"] = float(math.sin(2.0 * math.pi * hour / 24.0))
out["f_hour_cos"] = float(math.cos(2.0 * math.pi * hour / 24.0))
weekday = int(dt.weekday())
out["f_weekday"] = float(weekday)
label = session_label(hour)
out["f_sess_asian"] = 1.0 if label == "asian" else 0.0
out["f_sess_london"] = 1.0 if label == "london" else 0.0
out["f_sess_overlap"] = 1.0 if label == "overlap" else 0.0
out["f_sess_ny"] = 1.0 if label == "ny" else 0.0
out["f_minutes_to_hour"] = float((60 - int(dt.minute)) % 60)
out["f_day_of_month"] = float(int(dt.day))
out["f_first_friday"] = 1.0 if (weekday == 4 and int(dt.day) <= 7) else 0.0
def _fill_geometry(out: Dict[str, float], P: pd.DataFrame, setup,
atr: float, atr_ok: bool, norm, d: float) -> None:
out["f_atr"] = max(0.0, _finite(atr))
ranges = (P["High"] - P["Low"]).astype(float)
last_range = _finite(ranges.iloc[-1])
mean_range = _finite(ranges.mean())
out["f_atr_rel_500"] = (_finite(atr) / mean_range) if mean_range > 0 else 0.0
n = int(len(ranges))
out["f_range_pctile_500"] = (
float((ranges.to_numpy() < last_range).sum()) / n if n else 0.0)
out["f_bar_range_atr"] = norm(last_range)
high = _finite(P["High"].iloc[-1])
low = _finite(P["Low"].iloc[-1])
close = _finite(P["Close"].iloc[-1])
span = high - low
if span > 0:
out["f_close_pos_in_bar"] = (
(close - low) / span if d > 0 else (high - close) / span)
else:
out["f_close_pos_in_bar"] = 0.5
entry = _finite(getattr(setup, "entry_price", None))
stop = _finite(getattr(setup, "stop_loss", None))
tp1 = _finite(getattr(setup, "take_profit_1", None))
risk = abs(entry - stop)
out["f_risk_price"] = risk
out["f_sl_atr"] = norm(risk)
out["f_tp1_atr"] = norm(abs(tp1 - entry))
out["f_rr_planned"] = _finite(getattr(setup, "risk_reward", None))
def _fill_exhaustion(out: Dict[str, float], report, KL: Dict[str, float],
px: float, d: float, norm) -> None:
if d > 0:
swings = [KL[k] for k in ("swing_low_1", "swing_low_2", "swing_low_3")
if KL.get(k) is not None]
out["f_swing_dist_atr"] = norm(px - min(swings)) if swings else 0.0
else:
swings = [KL[k] for k in ("swing_high_1", "swing_high_2", "swing_high_3")
if KL.get(k) is not None]
out["f_swing_dist_atr"] = norm(max(swings) - px) if swings else 0.0
ema21 = KL.get("ema_medium")
out["f_ema21_dist_atr"] = (
norm(d * (px - _finite(ema21))) if ema21 is not None else 0.0)
ema200 = KL.get("ema_trend")
ema200_f = _finite(ema200, 0.0) if ema200 is not None else 0.0
if ema200 is not None and ema200_f != 0.0:
out["f_ema200_dist_atr"] = norm(d * (px - ema200_f))
out["f_ema200_dist_pct"] = d * (px - ema200_f) / ema200_f * 100.0
else:
out["f_ema200_dist_atr"] = 0.0
out["f_ema200_dist_pct"] = 0.0
out["f_exhaustion_count"] = _finite(getattr(report, "exhaustion_count", 0))
tags = getattr(report, "penalty_tags", None) or {}
for feature, tag in _TAG_FEATURES:
out[feature] = 1.0 if tag in tags else 0.0
out["f_penalty_total"] = _finite(getattr(report, "penalty_total", 0.0))
out["f_momentum_fading"] = 1.0 if getattr(report, "momentum_fading", False) else 0.0
def _fill_quality(out: Dict[str, float], report, setup, IND: dict,
d: float) -> None:
out["f_score_dir"] = d * _finite(getattr(report, "final_score", 0.0))
out["f_confidence"] = _finite(getattr(report, "final_confidence", 0.0))
out["f_quality_rank"] = _QUALITY_RANK.get(
getattr(report, "signal_quality", ""), 0.0)
agree = oppose = total = 0
for name, ind in (IND or {}).items():
if str(name).startswith("_"):
continue
if _finite(getattr(ind, "weight", 0.0)) == 0.0:
continue
total += 1
signal = getattr(ind, "signal", None)
if d > 0:
if signal in _BULLISH:
agree += 1
elif signal in _BEARISH:
oppose += 1
else:
if signal in _BEARISH:
agree += 1
elif signal in _BULLISH:
oppose += 1
out["f_ind_agree_frac"] = (agree / total) if total else 0.0
out["f_ind_oppose_frac"] = (oppose / total) if total else 0.0
out["f_is_buy"] = 1.0 if d > 0 else 0.0
strategy = str(getattr(setup, "strategy", "") or "")
# The ATR path stamps "XAU_ATR"; "XAU" is accepted for legacy setups.
out["f_strat_atr"] = 1.0 if strategy in ("XAU_ATR", "XAU") else 0.0
out["f_strat_pb"] = 1.0 if strategy == "XAU_PB" else 0.0
out["f_strat_cp"] = 1.0 if strategy == "XAU_CP" else 0.0
out["f_strat_generic"] = 1.0 if strategy == "GENERIC" else 0.0
def _fill_mtf(out: Dict[str, float], report, primary_tf: Timeframe,
confirmation_tfs: Sequence[Timeframe], d: float) -> None:
scores = getattr(report, "mtf_scores", None) or {}
out["f_mtf_primary"] = d * _finite(scores.get(primary_tf.value, 0.0))
confirms = list(confirmation_tfs or [])
for idx, key in ((0, "f_mtf_c1"), (1, "f_mtf_c2")):
if idx < len(confirms):
out[key] = d * _finite(scores.get(confirms[idx].value, 0.0))
else:
out[key] = 0.0
out["f_mtf_agreement"] = 1.0 if getattr(report, "mtf_agreement", False) else 0.0
def _fill_indicators(out: Dict[str, float], IND: dict, d: float, norm) -> None:
IND = IND or {}
def _val(name, default):
ind = IND.get(name)
return _finite(getattr(ind, "value", None), default) if ind else default
rsi = _val("rsi", 50.0)
out["f_rsi"] = rsi
out["f_rsi_dir"] = d * (rsi - 50.0)
bb_pos = _val("bollinger", 0.5) # bollinger.value = bb_position
out["f_bb_pos"] = bb_pos
out["f_bb_pos_dir"] = d * (bb_pos - 0.5)
out["f_adx"] = _val("adx", 0.0)
# keltner.value = close - kc_mid (signed extension from the mid line)
out["f_kc_ext_atr"] = norm(d * _val("keltner", 0.0))
out["f_roc_dir"] = d * _val("roc", 0.0)
# macd.value = histogram
out["f_macd_hist_dir"] = d * _val("macd", 0.0)
def _fill_path(out: Dict[str, float], P: pd.DataFrame, d: float, norm) -> None:
close = P["Close"].astype(float).to_numpy()
n = len(close)
for lag, key in ((6, "f_ret_6_atr"), (12, "f_ret_12_atr"),
(24, "f_ret_24_atr"), (48, "f_ret_48_atr")):
out[key] = norm(d * (close[-1] - close[-1 - lag])) if n >= lag + 1 else 0.0
run = 0.0
if n >= 2:
diffs = np.diff(close)
last = diffs[-1]
if last > 0:
sign = 1.0
elif last < 0:
sign = -1.0
else:
sign = 0.0
if sign != 0.0:
count = 0
for value in diffs[::-1]:
if (value > 0 and sign > 0) or (value < 0 and sign < 0):
count += 1
else:
break
run = sign * count
out["f_consec_closes_dir"] = max(-20.0, min(20.0, d * run))
out["f_bars_since_burst"] = float(BURST_LOOKBACK_BARS)
prev_atr = wilder_atr(P, 14).shift(1).to_numpy(dtype=float)
ranges = (P["High"] - P["Low"]).astype(float).to_numpy()
start = max(0, n - BURST_LOOKBACK_BARS)
for i in range(n - 1, start - 1, -1):
threshold = prev_atr[i]
if not math.isfinite(threshold) or threshold <= 0:
continue
if ranges[i] >= BURST_ATR_MULTIPLE * threshold:
out["f_bars_since_burst"] = float((n - 1) - i)
break
def _fill_cross_tf(out: Dict[str, float], dfs: Dict[Timeframe, pd.DataFrame],
confirmation_tfs: Sequence[Timeframe], d: float) -> None:
out["f_h1_ext_atr"] = _ext_vs_sma(dfs.get(Timeframe.H1), d)
out["f_c_other_ext_atr"] = 0.0
for tf in (confirmation_tfs or []):
if tf is Timeframe.H1:
continue
df = dfs.get(tf)
if df is None or len(df) < 21:
continue
out["f_c_other_ext_atr"] = _ext_vs_sma(df, d)
break
def build_features(
report,
setup,
dfs: Dict[Timeframe, pd.DataFrame],
primary_tf: Timeframe,
confirmation_tfs: Sequence[Timeframe],
) -> Dict[str, float]:
"""Decision-time feature vector for one signal.
Returns a dict whose keys are exactly ``FEATURE_ORDER`` (in order),
every value a finite Python float. Degenerate inputs fall back to
the neutral defaults instead of raising — this runs inside a scan.
"""
out: Dict[str, float] = dict(_DEFAULTS)
try:
P = dfs.get(primary_tf)
if P is None or len(P) == 0:
return {k: out[k] for k in FEATURE_ORDER}
P = P.iloc[-MAX_LOOKBACK_BARS:]
d = _direction(setup, report)
primary = getattr(report, "primary_analysis", None)
KL = dict(getattr(primary, "key_levels", None) or {})
IND = getattr(primary, "indicators", None) or {}
atr_raw = KL.get("atr")
atr = _finite(atr_raw, 0.0)
atr_ok = atr > 0
atr_div = atr if atr_ok else 1e-9
def norm(value: float) -> float:
"""ATR-normalise, or 0.0 when ATR is missing/non-positive."""
return (_finite(value) / atr_div) if atr_ok else 0.0
px = _finite(P["Close"].iloc[-1])
out["f_atr_pct"] = _finite(KL.get("atr_pct"), 0.0)
for step, args in (
(_fill_time, (out, decision_time(P, primary_tf))),
(_fill_geometry, (out, P, setup, atr, atr_ok, norm, d)),
(_fill_exhaustion, (out, report, KL, px, d, norm)),
(_fill_quality, (out, report, setup, IND, d)),
(_fill_mtf, (out, report, primary_tf, confirmation_tfs, d)),
(_fill_indicators, (out, IND, d, norm)),
(_fill_path, (out, P, d, norm)),
(_fill_cross_tf, (out, dfs, confirmation_tfs, d)),
):
try:
step(*args)
except Exception:
logger.debug("position_sizing: %s failed", step.__name__,
exc_info=True)
except Exception:
logger.debug("position_sizing.build_features failed", exc_info=True)
return {k: _finite(out.get(k), _DEFAULTS[k]) for k in FEATURE_ORDER}
# ══════════════════════════════════════════════════════════════════
# Stake multiplier
# ══════════════════════════════════════════════════════════════════
def stake_multiplier(
report,
setup,
dfs: Dict[Timeframe, pd.DataFrame],
primary_tf: Timeframe,
confirmation_tfs: Sequence[Timeframe],
artifact_dir=None,
) -> float:
"""Stake multiplier for a taken setup. Always ``1.0`` at this stage.
The guard order is already final — no artifact directory, a missing
artifact file, an artifact built for a different ``FEATURE_VERSION``,
or ANY exception all yield the neutral ``1.0``. Artifact evaluation
lands in a later step; until then this is wiring, not a decision.
"""
try:
if not artifact_dir:
return 1.0
artifact = load_artifact(artifact_path_for(primary_tf, artifact_dir))
if artifact is None:
return 1.0
if artifact.get("feature_version") != FEATURE_VERSION:
logger.debug(
"position_sizing: artifact feature_version %r != %d — neutral",
artifact.get("feature_version"), FEATURE_VERSION)
return 1.0
# Step 1: evaluation not implemented yet.
return 1.0
except Exception:
logger.debug("position_sizing.stake_multiplier failed", exc_info=True)
return 1.0