"""The shadow-portfolio step (issue #33, Slice 3e Part 4).
A forward simulator that consumes the decision *already written to
`decisions`* by the real pipeline, so every configured strategy — shadow or
trading — becomes measurable at a cost of zero extra model calls. Full
rationale: `docs/superpowers/specs/2026-08-11-shadow-portfolio-amended-design.md`.
**The isolation boundary, structural rather than conventional:**
- **This module never places an order.** Nothing here imports `brokers/` or
`execution/`. It takes bars, decisions, and a repository — no broker, no
`OrderExecutor`.
- **Simulated money never shares a table with real money.** No flag on
`trades` or `round_trips` marks a row as simulated; the five `sim_*`
tables are the only place simulated state exists, so a query over real
money is *incapable* of counting a simulated fill by construction.
- **A simulator failure must never stop the real cycle** — enforced by the
caller (`pipeline/run_once.py`), which wraps `advance()` in its own
`try`/`except`, the same belt-and-braces discipline `discover_symbols()`
gets from `cli/main.py`.
**Always-be-Buying (AbB), issue #42.** Every configured strategy also gets a
second, separately identified portfolio (`{strategy_id}::abb`,
`ABB_PORTFOLIO_SUFFIX` in `persistence/simulation.py`) that reads the exact
same recorded decisions but with `disable_sell_signals=True`: a `sell`
decision is ignored outright, so only the trailing stop can ever close a
position. It answers "how much of this strategy's return is the sell side
costing" in isolation from whether its buy/entry side is any good — see
`SimulationRunner.run()`, which steps both portfolios per strategy, each in
its own `try`/`except`.
"""
import json
import logging
import math
from collections.abc import Collection
from datetime import UTC, datetime, timedelta
from decimal import Decimal, InvalidOperation
from trader.config.schema import DiscoverySettings, PipelineConfig, SimulationSettings
from trader.domain import Account, Bar
from trader.persistence.bars import BarRepository, bar_width
from trader.persistence.decisions import DecisionRepository
from trader.persistence.models import Decision, SimPosition
from trader.persistence.simulation import (
EXIT_EVICTED,
EXIT_STRATEGY_SELL,
EXIT_TRAILING_STOP,
SimulationRepository,
)
from trader.risk.guardrails import REASON_DISCOVERY_CAP, ExposureCounter, Guardrails
__all__ = ["advance"]
_HUNDRED = Decimal(100)
_EPOCH = datetime(1970, 1, 1, tzinfo=UTC)
_logger = logging.getLogger("trader.simulation")
#: `inputs_json.veto_class` values the real pipeline writes — mirrored here
#: rather than imported, because importing `pipeline.run_once` would pull the
#: whole live pipeline (and its own imports) into `simulation/`. Two short
#: string constants are cheap to keep in sync by hand; re-check against
#: `pipeline/run_once.py`'s own `_VETO_CAPACITY`/`_VETO_SCREENING` if either
#: is ever renamed.
_VETO_CAPACITY = "capacity"
_VETO_SCREENING = "screening"
[docs]
def advance(
*,
strategy_id: str,
now: datetime,
bar_repository: BarRepository,
decision_repository: DecisionRepository,
simulation_repository: SimulationRepository,
settings: SimulationSettings,
pipeline_config: PipelineConfig,
discovery_settings: DiscoverySettings,
fixed_symbols: Collection[str],
portfolio_key: str | None = None,
disable_sell_signals: bool = False,
) -> None:
"""Step `strategy_id`'s simulated portfolio through every completed,
not-yet-applied bar of `settings.sim_interval`.
`portfolio_key` names the `SimPortfolio` row this call reads and writes;
it defaults to `strategy_id` when left unset, which is every caller
before issue #42. Decisions are **always** read using the real
`strategy_id` — only the portfolio/account identity changes — which is
how the Always-be-Buying (AbB) variant gets its own isolated ledger from
the same recorded decisions a strategy's normal portfolio already reads.
`disable_sell_signals` (issue #42) makes a recorded `signal_action ==
"sell"` a no-op for this call, as if it were unreadable — no round trip,
no cash or position change — so only the trailing-stop test can ever
close a position. It does not touch the stop itself.
Idempotent: a bar already recorded in `SimPortfolio.last_bar_ts` is never
reapplied, and `SimEquitySnapshot`'s own `(portfolio_id, bar_ts)` unique
constraint is the second, independent guard against the same thing.
Never raises past this function for anything that can be isolated to one
symbol or one bar — the caller's own `try`/`except` is the last resort,
not the first.
"""
interval = settings.sim_interval
width = bar_width(interval)
fixed = {s.strip().upper() for s in fixed_symbols}
portfolio_id_key = portfolio_key if portfolio_key is not None else strategy_id
portfolio = simulation_repository.get_or_create_portfolio(
portfolio_id_key, starting_cash=settings.sim_starting_cash
)
since = portfolio.last_bar_ts or _EPOCH
positions = {p.symbol: p for p in simulation_repository.all_positions(portfolio.id)}
pending = simulation_repository.pending_orders(portfolio.id)
relevant_symbols = set(positions) | {o.symbol for o in pending}
try:
decided_since = decision_repository.for_strategy_since(strategy_id, since)
except Exception: # noqa: BLE001 - a read failure must not stop other strategies
_logger.warning(
"simulation: could not read decisions for %s", strategy_id, exc_info=True
)
decided_since = []
relevant_symbols |= {d.ticker for d in decided_since}
if not relevant_symbols:
return
bar_timestamps: set[datetime] = set()
for symbol in relevant_symbols:
try:
bars = bar_repository.load_bars(symbol, interval, since, now)
except Exception: # noqa: BLE001 - one symbol's bar outage must not lose the rest
_logger.warning(
"simulation: could not load bars for %s", symbol, exc_info=True
)
continue
bar_timestamps.update(b.timestamp for b in bars)
completed = sorted(ts for ts in bar_timestamps if ts + width <= now)
cash = portfolio.cash
for bar_ts in completed:
try:
cash = _step_bar(
portfolio_id=portfolio.id,
strategy_id=strategy_id,
cash=cash,
bar_ts=bar_ts,
width=width,
interval=interval,
fixed=fixed,
bar_repository=bar_repository,
decision_repository=decision_repository,
simulation_repository=simulation_repository,
pipeline_config=pipeline_config,
discovery_settings=discovery_settings,
disable_sell_signals=disable_sell_signals,
)
except Exception: # noqa: BLE001 - one bad bar must not lose the rest of the backlog
_logger.error(
"simulation: step failed for %s at %s; stopping this strategy's "
"catch-up here so later bars are retried, not skipped",
strategy_id,
bar_ts,
exc_info=True,
)
return
simulation_repository.update_last_bar_ts(portfolio.id, bar_ts)
def _step_bar(
*,
portfolio_id: int,
strategy_id: str,
cash: Decimal,
bar_ts: datetime,
width: timedelta,
interval: str,
fixed: set[str],
bar_repository: BarRepository,
decision_repository: DecisionRepository,
simulation_repository: SimulationRepository,
pipeline_config: PipelineConfig,
discovery_settings: DiscoverySettings,
disable_sell_signals: bool = False,
) -> Decimal:
"""Mark, test the stop, fill, cancel, apply decisions, snapshot — for one
bar. Returns the portfolio's cash after every mutation this bar made.
Step order is load-bearing (design spec, amendment 3 / finding 5): fill
**before** cancel, or every pending order created at the end of a bar is
expired before it is ever tested for a fill, and the whole portfolio
silently sits at starting cash forever.
`disable_sell_signals` (issue #42) only touches the `signal_action ==
"sell"` branch inside step 5, below — the stop test in steps 1 & 2 is
unconditional and runs exactly the same whether this is set or not.
"""
positions = {p.symbol: p for p in simulation_repository.all_positions(portfolio_id)}
pending = {o.symbol: o for o in simulation_repository.pending_orders(portfolio_id)}
# Fetched here, ahead of steps 1-4, so a symbol with *no* existing
# position or pending order — a brand-new buy signal — still gets its
# bar loaded for step 5 to price the entry against. Symbols already
# tracked (open position, pending order) need their bar for steps 1-4
# regardless of whether they also have a decision this bar.
try:
span_decisions = decision_repository.for_strategy_since(
strategy_id, bar_ts, until=bar_ts + width
)
except Exception: # noqa: BLE001 - a read failure this bar must not lose the whole step
_logger.warning(
"simulation: could not read decisions for %s at %s",
strategy_id,
bar_ts,
exc_info=True,
)
span_decisions = []
latest_by_symbol: dict[str, Decision] = {}
for decision in span_decisions:
latest_by_symbol[decision.ticker] = decision # ascending order: last wins
symbols_this_bar = set(positions) | set(pending) | set(latest_by_symbol)
bars_at_ts: dict[str, Bar] = {}
for symbol in symbols_this_bar:
try:
found = bar_repository.load_bars(symbol, interval, bar_ts, bar_ts)
except Exception: # noqa: BLE001 - one symbol's bar outage must not lose the rest
_logger.warning(
"simulation: could not load the %s bar at %s",
symbol,
bar_ts,
exc_info=True,
)
continue
if found:
bars_at_ts[symbol] = found[0]
# --- 1 & 2: mark to market, then test the stop against the PRIOR mark ---
for symbol, position in list(positions.items()):
bar = bars_at_ts.get(symbol)
if bar is None:
continue
old_mark = position.high_water_mark
new_mark = max(old_mark, bar.high)
simulation_repository.mark_position(
position.id, high_water_mark=new_mark, last_price=bar.close
)
breach_price = old_mark * (1 - pipeline_config.trail_percent / _HUNDRED)
if bar.low <= breach_price:
realized_pl = (breach_price - position.entry_price) * position.quantity
simulation_repository.record_round_trip(
portfolio_id,
symbol,
decision_id=None,
quantity=position.quantity,
entry_price=position.entry_price,
exit_price=breach_price,
realized_pl=realized_pl,
opened_at=position.opened_at,
closed_at=bar_ts,
exit_reason=EXIT_TRAILING_STOP,
)
simulation_repository.close_position(position.id)
cash += breach_price * position.quantity
del positions[symbol]
else:
position.high_water_mark = new_mark
position.last_price = bar.close
# --- 3: fill pending orders where the bar's low reached the limit ---
for symbol, order in list(pending.items()):
bar = bars_at_ts.get(symbol)
if bar is None:
continue
if bar.low <= order.limit_price:
simulation_repository.fill_order(order.id)
# Issue #83: captured once, here at fill time, from the decision
# that proposed this order — `SimOrder.decision_id` already
# exists for exactly this "a fill's provenance survives" reason
# (see its own docstring). Never re-derived later: a position can
# outlive the bar range a later query would search, the same
# problem entry-time confidence solves on the real path.
entry_confidence = None
if order.decision_id is not None:
opening_decision = decision_repository.get(order.decision_id)
if opening_decision is not None:
entry_confidence = _confidence_of_decision(opening_decision)
position = simulation_repository.open_position(
portfolio_id,
symbol,
quantity=order.quantity,
entry_price=order.limit_price,
high_water_mark=order.limit_price,
last_price=order.limit_price,
opened_at=bar_ts,
entry_confidence=entry_confidence,
)
positions[symbol] = position
cash -= order.limit_price * order.quantity
del pending[symbol]
# --- 4: cancel anything still pending from a strictly earlier bar ---
for symbol, order in list(pending.items()):
if order.created_bar_ts < bar_ts:
simulation_repository.cancel_order(order.id)
del pending[symbol]
# --- 5: apply this bar's decisions, last row per symbol wins ---
# (`span_decisions`/`latest_by_symbol` were already fetched above, ahead
# of the bar-loading loop.)
exposure = ExposureCounter(
committed=sum((p.quantity * p.last_price for p in positions.values()), Decimal(0))
+ sum((o.quantity * o.limit_price for o in pending.values()), Decimal(0))
)
discovered_held = sum(
1 for symbol in (set(positions) | set(pending)) if symbol not in fixed
)
guardrails = Guardrails(pipeline_config)
# `confidence_ordering_enabled` (issue #86) mirrors
# `pipeline/run_once.py`'s allocation-order toggle: off, this is exactly
# today's iteration order (`latest_by_symbol`'s own dict order, i.e.
# whatever order `decision_repository.for_strategy_since` returned). On,
# every non-BUY decision (sells, holds, screening vetoes) is applied
# first, in that same original order — mirroring the real pipeline's own
# phase split, where every SELL executes in phase 1 before phase 2/3 even
# rank the BUY candidates — and only the BUY set is reordered, by
# `signal_confidence` descending (ties on symbol ascending, the same
# tiebreak `ranking.base.fallback_ranking` uses).
decision_items = list(latest_by_symbol.items())
if pipeline_config.confidence_ordering_enabled:
decision_items = _ordered_for_confidence_ordering(decision_items)
for symbol, decision in decision_items:
signal_action, veto_class = _parse_signal(decision)
if signal_action is None:
continue
if signal_action == "sell":
if disable_sell_signals:
# Always-be-Buying (issue #42): a sell signal is ignored,
# exactly like the `signal_action is None` case above — no
# round trip, no cash or position change. The trailing-stop
# test in steps 1 & 2 already ran this bar and is completely
# untouched by this flag; it remains the only exit.
continue
position = positions.get(symbol)
bar = bars_at_ts.get(symbol)
if position is None or bar is None:
continue
realized_pl = (bar.close - position.entry_price) * position.quantity
simulation_repository.record_round_trip(
portfolio_id,
symbol,
decision_id=decision.id,
quantity=position.quantity,
entry_price=position.entry_price,
exit_price=bar.close,
realized_pl=realized_pl,
opened_at=position.opened_at,
closed_at=bar_ts,
exit_reason=EXIT_STRATEGY_SELL,
)
simulation_repository.close_position(position.id)
cash += bar.close * position.quantity
del positions[symbol]
continue
if signal_action != "buy":
continue
if veto_class == _VETO_SCREENING:
# A screening veto is a fact about the symbol — untradable, a
# warrant, too cheap, too thin — equally true in simulation.
continue
# `capacity` vetoes (no headroom, the discovered cap, an existing
# pending entry the real pipeline already has) are ignored: they are
# facts about the *real* account, which this simulated portfolio does
# not share. It still refuses to pyramid — that is a structural rule
# of this app, not a capacity fact.
if symbol in positions or symbol in pending:
continue
bar = bars_at_ts.get(symbol)
if bar is None:
continue
limit_price = bar.close * (1 + Decimal(pipeline_config.limit_buffer_bps) / 10_000)
account = Account(
account_id=f"sim-{strategy_id}",
cash=cash,
equity=cash,
buying_power=cash,
portfolio_value=cash,
is_paper=True,
last_equity=cash,
)
cap = account.portfolio_value * pipeline_config.max_position_pct / _HUNDRED
if pipeline_config.confidence_sizing_enabled:
# Issue #85, Decision 1: mirrors `pipeline/run_once.py`'s
# `_fund` sizing formula by shape, not by import — the
# `simulation/` isolation rule (`CLAUDE.md`) forbids depending
# on `pipeline/`, the same reason `_confidence_of_decision`
# below is a hand-kept duplicate of `run_once.py`'s
# `_confidence_of` rather than an import of it.
cap = cap * _confidence_sizing_factor(
_confidence_of_decision(decision),
pipeline_config.confidence_sizing_floor,
)
if pipeline_config.volatility_sizing_enabled:
# Issue #88, Decision 4: composes multiplicatively with
# confidence sizing above when both are enabled, mirroring
# `pipeline/run_once.py`'s `_fund` by shape, not by import.
cap = cap * _volatility_sizing_factor(
_atr_pct_of_price_of_decision(decision),
pipeline_config.volatility_sizing_target_atr_pct,
)
proposed = max(int(cap // limit_price), 1) if limit_price > 0 else 0
if proposed <= 0:
continue
risk = guardrails.check_entry(
account,
symbol,
proposed,
limit_price,
is_discovered=symbol not in fixed,
discovered_positions_held=discovered_held,
max_discovered_positions=discovery_settings.max_discovered_positions,
exposure=exposure,
)
if not risk.allowed:
# Issue #83, gated behind `eviction_enabled` (`False` by
# default): a candidate capped out only by the discovered-
# position limit may still evict the weakest currently-held
# discovered *simulated* position instead of being turned away
# outright. Mirrors `pipeline/run_once.py`'s real-path eviction
# by shape, not by import — `simulation/` may never depend on
# `pipeline/` or `execution/`. Never funds `symbol` this same
# bar either way: see `_maybe_evict_for`.
if risk.code == REASON_DISCOVERY_CAP and pipeline_config.eviction_enabled:
cash = _maybe_evict_for(
symbol,
new_confidence=_confidence_of_decision(decision),
positions=positions,
fixed=fixed,
bars_at_ts=bars_at_ts,
cash=cash,
portfolio_id=portfolio_id,
bar_ts=bar_ts,
margin=pipeline_config.eviction_margin_confidence,
simulation_repository=simulation_repository,
)
continue
order = simulation_repository.create_order(
portfolio_id,
symbol,
decision_id=decision.id,
limit_price=limit_price,
quantity=Decimal(risk.approved_quantity),
created_bar_ts=bar_ts,
)
pending[symbol] = order
exposure.commit(limit_price * risk.approved_quantity)
if symbol not in fixed:
discovered_held += 1
simulation_repository.update_cash(portfolio_id, cash)
# --- 6: snapshot equity ---
equity = cash + sum(
(p.last_price * p.quantity for p in positions.values()), Decimal(0)
)
simulation_repository.snapshot_equity(portfolio_id, bar_ts, equity)
return cash
def _maybe_evict_for(
new_symbol: str,
*,
new_confidence: float | None,
positions: dict[str, SimPosition],
fixed: set[str],
bars_at_ts: dict[str, Bar],
cash: Decimal,
portfolio_id: int,
bar_ts: datetime,
margin: float,
simulation_repository: SimulationRepository,
) -> Decimal:
"""Evict the weakest currently-held discovered sim position (issue #83).
Local, hand-kept mirror of `pipeline/run_once.py`'s real-path eviction —
same shape (weakest *entry-time* confidence among known holders, strict
margin, never funds the new candidate this same step), not shared code:
`simulation/` may never import `pipeline/` or `execution/`.
Returns `cash`, updated only when an eviction actually happens —
`_step_bar`'s own "read a value, return the new one" convention, used
throughout this function rather than a nonlocal.
A holder with `entry_confidence is None` (a rule strategy's fill, or a
position opened before this field existed) is excluded from
consideration entirely, never treated as the weakest by some numeric
default — "unknown is not zero", the same rule `_DiscoveredHolder`
applies on the real path. `positions` is the *same* dict `_step_bar`
mutates throughout — deleting the evicted symbol from it here is what
stops a later decision in this same bar's loop (the evicted symbol's own
`sell`, if it has one) from acting on a position that is already gone,
exactly like the trailing-stop closes in steps 1 & 2 already rely on.
No fill price exists for the weakest holder at this bar (a thin listing,
a holiday) means no eviction this bar — never a fabricated exit price —
and the candidate stays rejected, to be retried whenever it (or a
stronger one) is next proposed.
"""
if new_confidence is None:
return cash
eligible = [
(symbol, position)
for symbol, position in positions.items()
if symbol not in fixed and position.entry_confidence is not None
]
if not eligible:
return cash
weakest_symbol, weakest = min(
eligible,
key=lambda item: item[1].entry_confidence, # type: ignore[arg-type,return-value]
)
# Strictly greater-than: a tie must never evict, or the margin is not a
# margin — same rule as the real path.
if new_confidence <= weakest.entry_confidence + margin: # type: ignore[operator]
return cash
bar = bars_at_ts.get(weakest_symbol)
if bar is None:
return cash
realized_pl = (bar.close - weakest.entry_price) * weakest.quantity
simulation_repository.record_round_trip(
portfolio_id,
weakest_symbol,
decision_id=None,
quantity=weakest.quantity,
entry_price=weakest.entry_price,
exit_price=bar.close,
realized_pl=realized_pl,
opened_at=weakest.opened_at,
closed_at=bar_ts,
exit_reason=EXIT_EVICTED,
)
simulation_repository.close_position(weakest.id)
cash += bar.close * weakest.quantity
del positions[weakest_symbol]
_logger.info(
"simulation: evicted %s (entry confidence %.3f) to free a discovered "
"slot for %s (confidence %.3f); %s is NOT funded this bar",
weakest_symbol,
weakest.entry_confidence,
new_symbol,
new_confidence,
new_symbol,
)
return cash
def _parse_signal(decision: Decision) -> tuple[str | None, str | None]:
"""`(signal_action, veto_class)` from `decision.inputs_json`.
`(None, None)` for anything unreadable — a missing blob, unparseable
JSON, a non-dict payload, or a row with no `signal_action` key (amendment
2 predates every row; an older one simply has nothing to read) — matching
this project's "everything LLM-adjacent degrades safely" discipline
rather than raising past a single bad row.
"""
if not decision.inputs_json:
return None, None
try:
inputs = json.loads(decision.inputs_json)
except ValueError:
return None, None
if not isinstance(inputs, dict):
return None, None
signal_action = inputs.get("signal_action")
if not isinstance(signal_action, str):
return None, None
veto_class = inputs.get("veto_class")
return signal_action, veto_class if isinstance(veto_class, str) else None
def _confidence_of_decision(decision: Decision) -> float | None:
"""`inputs_json["signal_confidence"]`, or `None` for anything unreadable.
Mirrors `pipeline/run_once.py`'s `_confidence_of` reading the same field
back out of a stored row rather than off a live `Signal` — not imported,
for the same reason `_parse_signal` above is a hand-kept mirror rather
than a `pipeline/` import: `simulation/` may never import `pipeline/` or
`execution/`. Also mirrors that sibling's `math.isfinite` guard (issue
#85's security review, 2026-08-30): `json.loads` accepts the
non-standard `NaN` literal, and an unguarded `NaN` fed into
`_confidence_sizing_factor`'s `max(0.0, min(1.0, x))` clamp evaluates to
`1.0` — a maximum sizing factor, not a rejected value. Every live writer
of `signal_confidence` already guards `math.isfinite` before persisting
(`_confidence_of`/`parse_decision`/`_to_confidence`), so this is not
reachable through this app's normal writers today, but this function
re-parses a persisted blob rather than trusting an upstream guarantee,
the same reasoning `_confidence_sizing_factor`'s own docstring gives for
its defensive `[0, 1]` clamp.
"""
if not decision.inputs_json:
return None
try:
inputs = json.loads(decision.inputs_json)
except ValueError:
return None
if not isinstance(inputs, dict):
return None
raw = inputs.get("signal_confidence")
if not isinstance(raw, int | float):
return None
value = float(raw)
return value if math.isfinite(value) else None
def _confidence_sizing_factor(confidence: float | None, floor: Decimal) -> Decimal:
"""`floor + (1 - floor) * confidence` — mirrors `pipeline/run_once.py`'s
identically-named function by shape, not by import; see that function's
docstring for the reasoning (already-approved BUYs are modulated, not
re-gated; `None` sizes at `floor`; the defensive `[0, 1]` clamp; why
`Decimal(str(...))` and not `Decimal(...)`)."""
if confidence is None:
return floor
clamped = max(0.0, min(1.0, confidence))
return floor + (1 - floor) * Decimal(str(clamped))
def _atr_pct_of_price_of_decision(decision: Decision) -> Decimal | None:
"""`inputs_json["technicals"]["atr_pct_of_price"]`, or `None`.
Mirrors `pipeline/run_once.py`'s `_atr_pct_of_price` reading the same
field back out of a stored row rather than off a live strategy instance
— not imported, for the same reason `_confidence_of_decision` above is a
hand-kept mirror rather than a `pipeline/` import: `simulation/` may
never import `pipeline/` or `execution/`. Only a decision recorded by
`LlmStrategy` carries a `technicals` key at all (a rule strategy's
`strategy_inputs` was always `{}`), so this is `None` for every
rule-strategy decision, `None` for a missing/unparseable value, and
never a fabricated reading — same "everything unreadable degrades
safely" discipline as `_confidence_of_decision` and `_parse_signal`
above.
"""
if not decision.inputs_json:
return None
try:
inputs = json.loads(decision.inputs_json)
except ValueError:
return None
if not isinstance(inputs, dict):
return None
technicals = inputs.get("technicals")
if not isinstance(technicals, dict):
return None
raw = technicals.get("atr_pct_of_price")
if raw is None:
return None
try:
return Decimal(str(raw))
except (InvalidOperation, ValueError, TypeError):
return None
def _volatility_sizing_factor(
atr_pct_of_price: Decimal | None, target_atr_pct: Decimal
) -> Decimal:
"""`min(1.0, target_atr_pct / atr_pct_of_price)` — mirrors
`pipeline/run_once.py`'s identically-named function by shape, not by
import; see that function's docstring for the full reasoning (shrink-only,
never inflates a calm name above `cap`; `None`/non-positive ATR sizes at
`1.0`, deliberately unlike `_confidence_sizing_factor`'s `floor` choice
for a missing confidence, because a missing ATR reading says nothing
about the symbol's volatility, only that no reading was computed)."""
if atr_pct_of_price is None or atr_pct_of_price <= 0:
return Decimal("1.0")
return min(Decimal("1.0"), target_atr_pct / atr_pct_of_price)
#: Sorts below any real confidence, including 0.0 — mirrors
#: `ranking.base`'s private `_ABSENT_CONFIDENCE`, kept as a local constant
#: for the same "cheap to keep in sync by hand" reason `_VETO_CAPACITY`/
#: `_VETO_SCREENING` above are.
_ABSENT_CONFIDENCE = -1.0
def _ordered_for_confidence_ordering(
items: list[tuple[str, Decision]],
) -> list[tuple[str, Decision]]:
"""Every non-BUY decision first (original order), then BUYs by confidence.
Used only when `pipeline_config.confidence_ordering_enabled` is set
(issue #86). A decision counts as a BUY here by the same `signal_action`
read `_step_bar`'s own loop uses one call later — this function does not
decide whether a BUY is actually actionable (already held, screening
veto, no bar); it only decides the *order* the main loop visits symbols
in, exactly like `run_once.py`'s ranking phase only orders approved
candidates and leaves every other decision to whatever already handles
it.
"""
buys: list[tuple[str, Decision, float]] = []
others: list[tuple[str, Decision]] = []
for symbol, decision in items:
action, _ = _parse_signal(decision)
if action == "buy":
confidence = _confidence_of_decision(decision)
buys.append(
(
symbol,
decision,
confidence if confidence is not None else _ABSENT_CONFIDENCE,
)
)
else:
others.append((symbol, decision))
buys.sort(key=lambda item: (-item[2], item[0]))
return others + [(symbol, decision) for symbol, decision, _ in buys]