"""One cycle of the trading pipeline — the service layer (requirements §4).
The CLI is a printer over this function; a dashboard or a future daemon calls
the same one. Nothing here knows about Typer, which is the whole point: the
Slice 1 carried debt warned that the `account` command inlined its
orchestration and that the daemon would copy that pattern.
Every collaborator is injected. That is not ceremony — it is what lets the
whole cycle be tested without a network, and what will let `OllamaEvaluator`
replace `BacktestEvaluator` next slice without this file changing.
"""
import json
import logging
import math
from collections.abc import Collection, Mapping, Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from decimal import Decimal, InvalidOperation
from trader.config.schema import PipelineConfig, StrategyMode
from trader.discovery.filters import FilterOutcome
from trader.domain import Bar, OrderSide, Position, SubmittedOrder
from trader.errors import TraderError
from trader.ranking.base import (
SOURCE_CONFIDENCE_ORDERING_ENABLED,
Candidate,
Ranker,
Ranking,
fallback_ranking,
)
from trader.ranking.comparables import build_comparables
from trader.risk.guardrails import (
REASON_DAILY_LOSS_HALT,
REASON_DISCOVERY_CAP,
ExposureCounter,
)
from trader.simulation import SimulationRunner
from trader.strategies.base import Action
__all__ = [
"CycleReport",
"StrategyRun",
"SymbolOutcome",
"reconcile_fills",
"run_once",
"settle_terminal_orders",
]
_HUNDRED = Decimal(100)
# How far before the oldest outstanding submission the fill poll reaches back.
# Absorbs clock skew between this app's recorded `submitted_at` and the one
# Alpaca filters `after` on. See `_fill_poll_since`.
_FILL_POLL_OVERLAP = timedelta(hours=1)
# Broker statuses that mean "this order is over and it will never fill again".
# Every one of them is a *terminal* state at Alpaca, which is what makes the
# answer durable: a settled row is invisible to `reconcile_fills` forever, so a
# status that could still progress to a fill must never appear here.
#
# `replaced` belongs: Alpaca's replace is a cancel-and-resubmit, so the replaced
# id is dead and its shares (if any) live on the new order. `ratchet_protection`
# already settles that row explicitly at the point of the replace; this catches
# the one whose settle did not happen — a replace made outside this app, or a
# cycle that died between the replace and the stamp.
#
# Deliberately absent: `done_for_day` (a GTC order resumes tomorrow),
# `pending_cancel` and `pending_replace` (both can still fill in the meantime),
# `suspended`, `stopped` and `calculated`. Anything not listed is left
# unsettled, which costs a stale fill-poll window — bounded, and already
# tracked — rather than a lost fill, which is not.
_TERMINAL_NON_FILLING_STATUSES = frozenset(
{
"canceled",
# Alpaca spells it with one `l`. The other spelling is here because a
# settle must not silently stop happening if the SDK ever normalises it,
# and both readings mean the same terminal thing.
"cancelled",
"expired",
"rejected",
"replaced",
}
)
# How many per-order broker lookups the terminal-order sweep may make in one
# cycle. A bound rather than "all of them" because the row count is unbounded in
# principle and a cycle's broker budget is not; oldest-first ordering means the
# spend always goes where it narrows the fill-poll window, and whatever is left
# over is picked up next cycle.
_MAX_SETTLE_LOOKUPS = 25
# The class of thing that overwrote a strategy's action, recorded as
# `inputs_json.veto_class`. Two values, and the split is what lets a reader
# decide whether the veto applies to *it*:
#
# `capacity` — a fact about this account's money or its holdings: no headroom
# under either cap, the discovered-position limit, already long,
# an entry order already pending, the daily loss limit, or losing
# a same-symbol collision. A portfolio with its own cash does not
# inherit any of these.
# `screening` — a fact about the *symbol* or the signal's history on it:
# untradable, a warrant, too cheap, too thin, or the backtest
# evaluator declining. True regardless of whose money is at risk.
#
# Absent means nothing vetoed the decision. Do not collapse these into one value
# in either direction: the shadow portfolio ignores `capacity` and honours
# `screening`, so a single class would make it either refuse to measure a
# strategy against capital it never had, or buy untradable warrants in
# simulation.
_VETO_CAPACITY = "capacity"
_VETO_SCREENING = "screening"
# `decisions.rejection_reason` (issue #21): which specific gate produced a
# non-buy/non-sell final action when the strategy's own `signal_action`
# (captured above every veto) said otherwise. `veto_class` above already
# answers "capacity or screening"; this answers "which one, specifically" —
# a diagnostic run against a copied production DB in 2026-08-17 had to
# reconstruct that by symbol-matching and reading free-text `reasoning`,
# which breaks the moment a message is reworded. Set alongside `veto`
# everywhere `veto` is set, and never derived from `reason`'s prose, for the
# same reason `veto_class` itself is not: wording is free to change, a stored
# code must not silently change meaning under it.
#
# The `discovery.filters.REASON_*` constants (price_floor, volume_floor,
# not_tradable, not_common_stock, analyst_net_negative, no_price_history) and
# `risk.guardrails.REASON_*` constants (daily_loss_halt, discovery_cap,
# invalid_price, total_exposure_cap, position_cap) cover the gates that live
# outside this module; the ones below are this module's own.
#
# Deliberately absent: `fingerprint_reuse`. The news-fingerprint gate
# (`LlmStrategy._reuse_or_none`) only ever carries a HOLD forward, so a reused
# row's `signal_action` is already "hold" — identical to the recorded action,
# not a divergence this column exists to name. That mechanism already has its
# own structured fields (`reused_model_reason`, `reuse_origin_decision_id`) in
# `inputs_json`; duplicating them here would be a second definition of the
# same fact, free to drift from the first.
_REJECTION_ALREADY_EXITED = "already_exited"
_REJECTION_ALREADY_LONG = "already_long"
_REJECTION_PENDING_ENTRY = "pending_entry"
_REJECTION_EVALUATOR = "evaluator_rejected"
_REJECTION_DUPLICATE_APPROVAL = "duplicate_approval"
_logger = logging.getLogger("trader.pipeline")
[docs]
@dataclass(frozen=True, slots=True)
class StrategyRun:
"""One configured strategy, its mode, and the evaluator that gates it.
Bundled rather than passed as three parallel sequences, because three
lists that must stay index-aligned is a bug waiting to happen.
"""
strategy: object
mode: StrategyMode
evaluator: object | None = None
@property
def is_live(self) -> bool:
"""Whether this strategy may actually place orders."""
return self.mode is StrategyMode.TRADING
@dataclass(slots=True)
class _DiscoveredCapCounter:
"""A mutable running count of discovered symbols committed this cycle.
Seeded once, before the symbol loop, from current positions plus any
already-open pending entry orders — then incremented in place inside
`_allocate`'s loop every time *this* cycle submits a discovered entry, so
the next discovered candidate in the same cycle sees the higher count rather
than a pre-loop snapshot.
A plain `int` computed once before the loop and never touched again is
exactly what let six discovered BUY signals against a cap of two place
six limit buys in a single cycle: the cap only ever compared against what
existed *before* the cycle started, never against what the cycle itself
was doing. `discovered_positions_held` from current positions alone has
the same blind spot this app's own duplicate-order bug had — "a position
is not the whole picture; check open orders too" — so the seed counts
pending entries as well, via the same `pending_entry_for` a single-symbol
check already uses.
"""
held: int
@dataclass(frozen=True, slots=True)
class _DiscoveredHolder:
"""A currently-held discovered position, eligible to be evicted (issue #83).
Built once per cycle, only when `pipeline_config.eviction_enabled`, from
`positions_discovered` — never from `symbols`, for the same reason
`discovered_positions_held` itself is not: a position whose news story
faded and left the scan is still a real holding occupying a real slot,
and would not even be re-evaluated this cycle if `symbols` were the
source. That is exactly why `entry_confidence` is read from the decision
that *opened* the position rather than from "this cycle's fresh
confidence" — a fresh re-evaluation may not exist at all for a symbol
outside this cycle's scan, while the opening decision always does.
`entry_confidence` and `strategy_id` are `None` together or not at all:
both come from the same lookup (`open_round_trip_for` -> its
`decision_id` -> that decision's `inputs_json.signal_confidence`), and a
holder with either missing is excluded from eviction consideration
entirely by `_eligible_eviction_targets` — "unknown is not zero" applies
here exactly as it does to a missing analyst opinion: a holder this app
cannot score is never the weakest, because it might not be.
"""
symbol: str
position: Position
entry_confidence: float | None
strategy_id: str | None
[docs]
@dataclass(frozen=True, slots=True)
class SymbolOutcome:
"""What one strategy decided about one symbol, and why."""
symbol: str
strategy_id: str
action: str
reason: str
trade_id: int | None = None
order_id: str | None = None
#: 1-based place in the ranking that funded this cycle's entries, or
#: `None` for an outcome that never competed for capital. Surfaced so a
#: working ranker and a silently failing one do not print identically.
rank: int | None = None
#: The specific gate that overrode the strategy's own signal, when one
#: did — the same value written to `decisions.rejection_reason` (issue
#: #21). `None` whenever nothing overrode the signal, mirroring `veto`
#: in `_record_decision`; it is not derived from `reason`'s free text.
rejection_reason: str | None = None
@dataclass(frozen=True, slots=True)
class _DeferredEntry:
"""An approved BUY waiting on capital.
Everything that does not depend on available capital has already been
checked: the strategy is `trading`, there is no position, no unfilled entry
order, no entry screen, the cycle is not halted, and the evaluator (if any)
approved. What remains is whether there is room.
`strategy_inputs` is a **snapshot**, not a live read.
`LlmStrategy.last_inputs` is per-instance mutable state and one instance
serves every symbol in a cycle, so reading it during allocation would
attach the *last* symbol's news and `prompt_sha256` to every row. That is
the bug `_process_symbol` already documents, and the phase split reopens it
unless the snapshot travels with the candidate. A wrong-symbol prompt hash
is still a *valid* hash, so the acceptance check that rebuilds the prompt
from `inputs_json` would report MATCH on a mislabelled row.
"""
symbol: str
run: StrategyRun
signal: object
last_close: Decimal
evaluator_reason: str
evidence: dict[str, object]
strategy_inputs: dict[str, object]
is_discovered: bool
candidate: Candidate
[docs]
@dataclass(frozen=True, slots=True)
class CycleReport:
"""Everything one cycle did, for the CLI to print or a caller to inspect."""
outcomes: list[SymbolOutcome] = field(default_factory=list)
stops_placed: list[str] = field(default_factory=list)
#: Symbols whose fixed protective stop was raised this cycle. Populated
#: from `OrderExecutor.ratchet_protection`, kept distinct from
#: `stops_placed` because a ratchet replaces an existing stop rather than
#: placing a new one, and the two must never be conflated in a report an
#: operator reads.
stops_ratcheted: list[str] = field(default_factory=list)
#: Symbols whose own unfilled entry BUY was cancelled by this app because
#: the close was near. Populated from `OrderExecutor.cancel_stale_entries`
#: — see `_cancel_entries_near_close`'s docstring for why this exists:
#: the entry became GTC to fix issue #3, and nothing else expires it.
entries_cancelled: list[str] = field(default_factory=list)
#: Broker order ids whose `trades` row was settled this cycle because the
#: order ended terminally without ever filling — cancelled, expired,
#: rejected or replaced with a zero fill. Issue #1: nothing else ever
#: settles those rows, and each one holds the fill-poll window open at its
#: own submission time until it is. Populated by `settle_terminal_orders`.
orders_settled: list[str] = field(default_factory=list)
snapshot_id: int | None = None
halted: bool = False
#: Symbols holding a position with no open protective SELL on record, as
#: of the very end of this cycle — after reconciliation has already had
#: its turn. Requirements §8: "the app must never terminate quietly while
#: it believes money is unprotected." Populated by
#: `_check_end_of_cycle_protection`; empty means "checked and found
#: nothing", never "not checked".
unprotected: list[str] = field(default_factory=list)
#: Symbols holding a position with **more than one** open protective SELL
#: at the end of this cycle. The mirror of `unprotected`, and the worse of
#: the two failures: two sell claims on the same shares can both execute,
#: and on a long-only account the second one opens a short — an
#: unbounded-loss position this app has no guardrails for. Checked from the
#: same broker-wide open-orders read as `outstanding_entries`, so it costs
#: nothing extra.
double_protected: list[str] = field(default_factory=list)
#: Symbols with an unfilled entry BUY still open at the broker at the end
#: of this cycle. On 2026-08-07 a daemon cycle submitted two buys, logged
#: a clean summary, and exited 0 while both sat unfilled — nothing
#: checked, so nothing warned. This is what makes that loud instead,
#: whatever process outlives (or doesn't outlive) the order.
outstanding_entries: list[str] = field(default_factory=list)
[docs]
def run_once(
*,
broker,
cache,
strategies: Sequence[StrategyRun],
guardrails,
executor,
trade_repository,
decision_repository,
snapshot_repository,
outcome_repository,
symbols: Sequence[str],
pipeline_config: PipelineConfig,
now: datetime | None = None,
previous_positions: dict[str, Position] | None = None,
fixed_symbols: Collection[str] | None = None,
max_discovered_positions: int = 0,
entry_blocked: Mapping[str, FilterOutcome] | None = None,
ranker: Ranker | None = None,
simulation: SimulationRunner | None = None,
benchmark_symbol: str | None = None,
) -> CycleReport:
"""Walk the watchlist once with every configured strategy.
Every strategy sees every symbol, so the report grows from one entry per
symbol to one per (symbol, strategy) pair. Only the strategy configured as
`live` may reach the executor; every other one is shadowed — evaluated and
recorded, but never allowed to spend money.
Three phases, in this order:
1. **Decide.** Every (symbol, strategy) pair is evaluated and gets a
`decisions` row. SELLs execute *here*, inside this phase, because they
are time-sensitive and do not compete for capital — deferring one behind
a ranking call would put a signal-driven exit behind a model call. So do
holds, halts, entry screens, evaluator rejections and errors. An
approved BUY submits nothing; it becomes a `_DeferredEntry` instead.
2. **Rank.** The deferred entries are ordered by `ranker`, or by the
deterministic confidence fallback when there is none. Never raises: a
ranking failure degrades the order, it does not cost the cycle.
`pipeline_config.confidence_ordering_enabled` (issue #86, off by
default) forces the confidence-descending order regardless of
`ranker`, to isolate that mechanism from LLM ranking when simulating.
3. **Allocate.** The ranked candidates are funded top-down until the
headroom runs out, with `ExposureCounter` and the discovered cap still
advancing inside the loop. Whichever candidate sits earliest in the
watchlist no longer wins the capital by virtue of coming first.
"""
decided_at = now or datetime.now(UTC)
# First, so a fill that happened while nothing was watching is on record
# before this cycle reasons about anything. This ordering is itself the
# fix for the duplicate-order class of bug: an in-flight fill must be
# accounted for before pending-order and position checks run against it.
reconcile_fills(
broker=broker,
outcome_repository=outcome_repository,
trade_repository=trade_repository,
previous_positions=previous_positions
or {p.symbol: p for p in broker.get_positions()},
now=decided_at,
)
# Then settle whatever ended without ever filling, which is the one outcome
# `reconcile_fills` structurally cannot learn about (issue #1). *After* it,
# not before: a fill this cycle just reconciled is already settled by the
# round-trip commit, so the sweep asks the broker about strictly fewer
# orders. It cannot go the other way round and be cheaper, and running it
# first would narrow this cycle's own poll window by at most one cycle's
# worth of staleness — not worth the extra lookups.
orders_settled = settle_terminal_orders(
broker=broker, trade_repository=trade_repository, now=decided_at
)
account = broker.get_account()
positions_by_symbol = {p.symbol: p for p in broker.get_positions()}
halted = guardrails.daily_loss_breached(account)
if halted:
_logger.warning(
"daily loss limit breached: equity %s against %s at yesterday's close; "
"no new entries this cycle",
account.equity,
account.last_equity,
)
# Cancel this app's own stale unfilled entries before anything below
# reads exposure or open orders, so a cancellation frees its headroom for
# this same cycle rather than the next one. Broker-wide, not scoped to
# this cycle's own `symbols` — see `_cancel_entries_near_close`'s
# docstring for why that scoping would silently miss exactly the
# symbols this app most needs to sweep.
entries_cancelled = _cancel_entries_near_close(
executor=executor,
broker=broker,
trade_repository=trade_repository,
pipeline_config=pipeline_config,
decided_at=decided_at,
)
# A position counts as discovered when its symbol is not one the operator
# listed — NOT when it is in the current scan. A story fades, the symbol
# leaves the scan, and the position remains; counting only scanned symbols
# would let the cap be exceeded by exactly those stale positions.
#
# `fixed_symbols is None` ("not passed") and `fixed_symbols == []`
# ("explicitly empty") must be told apart: a pure-discovery configuration
# legitimately has no fixed tickers at all, and `... or symbols` would
# treat that empty list the same as "not passed" and fall back to
# `symbols` — silently treating the entire scanned universe as fixed, so
# `is_discovered` is never True and the cap never engages.
fixed = {
s.strip().upper() for s in (symbols if fixed_symbols is None else fixed_symbols)
}
positions_discovered = {
p.symbol.upper()
for p in positions_by_symbol.values()
if p.symbol.upper() not in fixed
}
# A discovered symbol with an unfilled entry order is not a position, but
# it is exposure already committed — the same reasoning `pending_entry_for`
# exists for at all. Checked per symbol, scoped to this cycle's own scan,
# with `executor.pending_entry_for` (the one mechanism this app already
# uses to ask the broker this question), never a second bespoke broker
# read. A symbol already counted via a position is skipped: this app does
# not pyramid, so a symbol never legitimately carries both at once, and a
# stale open order behind an already-closed position must not be
# double-counted against the one symbol.
pending_discovered = {
symbol.upper()
for symbol in symbols
if symbol.upper() not in fixed
and symbol.upper() not in positions_by_symbol
and executor.pending_entry_for(symbol) is not None
}
discovered_cap = _DiscoveredCapCounter(len(positions_discovered | pending_discovered))
# Built only when the flag can act on it: every entry costs two extra
# reads (`open_round_trip_for`, then `decision_repository.get`), and an
# unused list would just be spending them for nothing while
# `eviction_enabled` is `False`, which it is by default (issue #83).
discovered_holders: list[_DiscoveredHolder] = (
_build_discovered_holders(
positions_discovered=positions_discovered,
positions_by_symbol=positions_by_symbol,
outcome_repository=outcome_repository,
decision_repository=decision_repository,
)
if pipeline_config.eviction_enabled
else []
)
# Capital already committed, before this cycle spends anything: open
# positions at market value, plus entry orders placed but not yet filled.
# A pending limit buy is money spoken for even though no position exists
# yet — the same reasoning `pending_entry_for` exists for at all.
#
# One unfiltered `get_open_orders()` rather than one call per symbol: it
# is a single round trip instead of N, and it counts a pending buy for a
# symbol that has since dropped out of the scan, which a per-symbol loop
# over `symbols` would miss entirely.
committed = sum((p.market_value for p in positions_by_symbol.values()), Decimal(0))
try:
for order in broker.get_open_orders():
if (
order.side is OrderSide.BUY
and order.limit_price is not None
and order.symbol.upper() not in positions_by_symbol
):
committed += order.limit_price * order.quantity
except TraderError as exc:
# Seeding high is the safe direction, but we cannot know the true
# figure — so refuse to spend at all this cycle rather than guess low.
_logger.error(
"could not read open orders for the exposure seed: %s; "
"treating the account as fully committed for this cycle",
exc,
)
committed = account.portfolio_value * _HUNDRED
exposure = ExposureCounter(committed)
# Fetched once per cycle, not once per symbol: every symbol shares the
# same decision window, so N `cache.ensure(benchmark_symbol, ...)` calls
# would be N redundant lookups of the same span (issue #87). `None`
# symbol means "not configured" (`run_once`'s own default), and a fetch
# failure degrades to `None` bars rather than raising — same "advisory
# input, must not cost the cycle" discipline as discovery and the
# analyst/news fetches inside `LlmStrategy`. `SPY` (the default) is
# already a `fixed_tickers` entry on this project's live watchlist, so
# by the time its own symbol loop iteration runs this is usually already
# cached — but this call may still be the one that populates it, if
# nothing earlier this cycle has touched it yet.
benchmark_bars: list[Bar] | None = None
if benchmark_symbol:
try:
benchmark_start = decided_at - timedelta(
days=pipeline_config.evaluation_lookback_days
)
benchmark_bars = cache.ensure(benchmark_symbol, benchmark_start, decided_at)
except TraderError as exc:
_logger.warning(
"relative-strength benchmark %s unavailable this cycle: %s",
benchmark_symbol,
exc,
)
benchmark_bars = None
# Phase 1. Every pair decides and is recorded; approved BUYs are deferred
# rather than submitted, so no headroom is spent before the ranking runs.
outcomes: list[SymbolOutcome] = []
deferred: list[_DeferredEntry] = []
# Symbol -> the strategy that has already exited it this cycle. Empty at the
# start of every cycle, because a position sold last cycle is gone and a
# position still held is fair game again. See `_process_symbol`.
sold_this_cycle: dict[str, str] = {}
for symbol in symbols:
for run in strategies:
result = _process_symbol(
symbol=symbol,
run=run,
account=account,
position=positions_by_symbol.get(symbol),
halted=halted,
cache=cache,
executor=executor,
trade_repository=trade_repository,
decision_repository=decision_repository,
pipeline_config=pipeline_config,
decided_at=decided_at,
is_paper=account.is_paper,
sold_this_cycle=sold_this_cycle,
is_discovered=symbol.upper() not in fixed,
entry_blocked=entry_blocked,
benchmark_bars=benchmark_bars,
benchmark_symbol=benchmark_symbol,
)
if isinstance(result, _DeferredEntry):
deferred.append(result)
else:
outcomes.append(result)
# Phase 2. Nothing above submitted an entry, so no headroom has been spent
# by the time the ranking decides who gets it.
outcomes.extend(
_allocate(
candidates=deferred,
ranker=ranker,
account=account,
guardrails=guardrails,
executor=executor,
trade_repository=trade_repository,
decision_repository=decision_repository,
outcome_repository=outcome_repository,
pipeline_config=pipeline_config,
decided_at=decided_at,
is_paper=account.is_paper,
discovered_cap=discovered_cap,
max_discovered_positions=max_discovered_positions,
exposure=exposure,
discovered_holders=discovered_holders,
sold_this_cycle=sold_this_cycle,
)
)
# Reconciliation runs regardless of the halt: §8 halts new *entries*, not
# protection. Positions are re-read so anything closed above is gone.
stops_placed: list[str] = []
stops_ratcheted: list[str] = []
try:
current: list[Position] = broker.get_positions()
# Ratchet FIRST, reconcile SECOND. This order is load-bearing and was
# the other way round until the whole-branch review measured what
# happens when a ratchet fails: Alpaca's replace is a
# cancel-and-resubmit under the hood, so `replace_stop_price` can raise
# *after* the old stop is already gone, leaving the position with no
# protective order at all — and with reconcile already finished for the
# cycle, nothing repaired it until the next one. On `run-once`, or the
# last cycle of `--max-cycles`, there is no next one. Ratcheting first
# costs nothing, because `ratchet_protection` skips a position with no
# protective order anyway (that has always been reconcile's job), and it
# puts the repair *after* the only step that can open the hole: a failed
# replace leaves no visible protective SELL, so `reconcile_protection`
# sees an unprotected position in this same cycle and places a fresh
# trailing stop on it.
#
# `replace_stop_price` mints a brand-new broker order id (Alpaca's
# replace is a cancel-and-resubmit under the hood, confirmed against
# the docs 2026-08-10: the old order transitions to `replaced` and the
# response is a distinct order), so the replacement is recorded here
# exactly like a freshly placed stop — for the same reason: an
# unrecorded fill is a fill `reconcile_fills` can never attribute to
# this app, and a ratcheted stop is still this position's *only* exit
# path.
#
# The row for the order each ratchet *replaced* must also be settled,
# or it pins `oldest_unsettled_submitted_at()` (issue #1) at its
# submission time forever — and unlike the rare cancelled/expired
# order issue #1 already tracks, this would fire on every ratchet.
# `RatchetedStop.replaced_order_id` is what makes that possible:
# `ratchet_protection` reads the replaced order at the point it looks
# up the existing protective order, so the id was never actually
# unavailable — only the old (bare `SubmittedOrder`) return type threw
# it away before it reached here. `mark_settled` no-ops for an id with
# no matching row (a hand-placed stop, say) rather than inventing one.
for ratcheted in executor.ratchet_protection(current):
_record_protective_order(
ratcheted.order,
trade_repository=trade_repository,
outcome_repository=outcome_repository,
is_paper=account.is_paper,
)
trade_repository.mark_settled([ratcheted.replaced_order_id], decided_at)
stops_ratcheted.append(ratcheted.order.symbol)
# Then protect anything left without a visible stop — including any
# position whose ratchet just failed with the old stop already
# cancelled. Same `current` snapshot on purpose: neither loop mutates
# it, and re-reading positions between the two would cost a broker call
# to learn nothing this cycle acts on.
#
# Recording the stop is not bookkeeping — it is what makes its
# eventual fill recognisable. `reconcile_fills` ignores any fill
# with no `trades` row as "not ours", and a protective stop is this
# app's *primary* exit: every position gets one, whether or not a
# strategy ever emits a SELL. Unrecorded, a stop-driven exit left
# its round trip open forever, and later unrelated sells then
# closed it against a stale entry price — a fabricated realized
# P/L, not merely a missing one.
for order in executor.reconcile_protection(current):
_record_protective_order(
order,
trade_repository=trade_repository,
outcome_repository=outcome_repository,
is_paper=account.is_paper,
)
stops_placed.append(order.symbol)
except TraderError as exc:
_logger.error("reconciliation failed: %s", exc)
# Last, and independent of whether reconciliation above succeeded: what is
# this cycle actually handing back? See the function's own docstring for
# why this is never allowed to conclude "nothing to report" from a broker
# read it could not make.
unprotected, double_protected, outstanding_entries = _check_end_of_cycle_protection(
broker=broker
)
snapshot_id: int | None = None
try:
snapshot_id = snapshot_repository.save_snapshot(
broker.get_account(), broker.get_positions(), captured_at=decided_at
)
except Exception as exc: # noqa: BLE001 - a snapshot failure must not lose the cycle
_logger.error("could not save snapshot: %s", exc)
# The shadow-portfolio simulator (issue #33) is a measurement instrument,
# never a trading path — its failure must never cost this cycle's real
# report, the same discipline the snapshot save above uses.
if simulation is not None:
try:
simulation.run(
strategy_ids=[run.strategy.id for run in strategies], # type: ignore[attr-defined]
now=decided_at,
)
except Exception as exc: # noqa: BLE001 - never lose the cycle to this
_logger.error("simulation run failed: %s", exc)
return CycleReport(
outcomes=outcomes,
stops_placed=stops_placed,
stops_ratcheted=stops_ratcheted,
entries_cancelled=entries_cancelled,
orders_settled=orders_settled,
snapshot_id=snapshot_id,
halted=halted,
unprotected=unprotected,
double_protected=double_protected,
outstanding_entries=outstanding_entries,
)
def _entry_confidence_of(decision) -> float | None:
"""`decisions.inputs_json["signal_confidence"]` off a stored `Decision`.
The entry-time mirror of `_confidence_of`, which reads the same field off
a live `Signal` before it is ever persisted. Never raises: a missing
blob, unparseable JSON, a non-dict payload, or a non-numeric value all
degrade to `None` — "I could not tell" — the same discipline
`DecisionRepository.latest_decision` already applies to this exact
column for the same reason (a reader here, `_build_discovered_holders`,
must never *guess* a confidence low or high; a holder it cannot score is
excluded from eviction consideration entirely, never treated as the
weakest by default).
"""
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 isinstance(raw, bool) or not isinstance(raw, (int, float, Decimal)):
return None
value = float(raw)
return value if math.isfinite(value) else None
def _build_discovered_holders(
*,
positions_discovered: Collection[str],
positions_by_symbol: Mapping[str, Position],
outcome_repository,
decision_repository,
) -> list[_DiscoveredHolder]:
"""One `_DiscoveredHolder` per currently-held discovered symbol (issue #83).
`entry_confidence` is read from the decision that *opened* the position —
`outcome_repository.open_round_trip_for(symbol)` for the open trip, then
`decision_repository.get(trip.decision_id)` for the row itself — never
from a fresh re-evaluation this cycle. Two reasons, both load-bearing:
1. A held discovered symbol whose news story has faded may not even be in
this cycle's `symbols` (see `positions_discovered`'s own comment in
`run_once`), so "this cycle's decision" may not exist at all for it.
2. Even when it does exist, more than one `mode: trading` strategy can
evaluate the same held symbol in one cycle (this project's own "more
than one strategy may trade" rule) and only one of them owns the
position — picking the wrong one's confidence would compare the new
candidate against an opinion nobody staked the position on.
A holder whose entry decision, `decision_id`, or `signal_confidence`
cannot be found is still included in the returned list, but with
`entry_confidence=None` — excluded from eviction consideration by
whichever caller filters on that (`_fund`), never defaulted to a
sentinel that could read as "weakest" or "strongest" by accident.
Never raises: one symbol's lookup failing must not cost every other
holder's eligibility, the same discipline every other advisory read in
this module already uses.
"""
holders: list[_DiscoveredHolder] = []
for symbol in positions_discovered:
position = positions_by_symbol.get(symbol)
if position is None:
# Case-folding mismatch between the two maps' keys should not
# happen (both are built from the same broker read), but a
# holder this function cannot even locate a Position for cannot
# be evicted, so it is skipped rather than passed through with a
# placeholder.
continue
entry_confidence: float | None = None
strategy_id: str | None = None
try:
trip = outcome_repository.open_round_trip_for(symbol)
if trip is not None and trip.decision_id is not None:
decision = decision_repository.get(trip.decision_id)
if decision is not None:
entry_confidence = _entry_confidence_of(decision)
# Both come from the same lookup, and are excluded
# together below when either is missing — see the
# class docstring.
if entry_confidence is not None:
strategy_id = trip.strategy_id
except Exception as exc: # noqa: BLE001 - advisory; must not cost the cycle
_logger.warning(
"could not read the entry confidence for held discovered "
"position %s: %s; excluding it from eviction consideration "
"this cycle",
symbol,
exc,
)
holders.append(
_DiscoveredHolder(
symbol=symbol,
position=position,
entry_confidence=entry_confidence,
strategy_id=strategy_id if entry_confidence is not None else None,
)
)
return holders
def _eligible_eviction_targets(
holders: Sequence[_DiscoveredHolder], sold_this_cycle: Mapping[str, str]
) -> list[_DiscoveredHolder]:
"""Holders this cycle may still evict: a known score, not already exited.
`sold_this_cycle` guards against the double-sell this project has a hard
rule against: a holder whose own strategy already sold it in phase 1
(a signal-driven exit, not a cap issue) must never be sold *again* here
— `OrderExecutor.exit_position` would try to cancel an already-cancelled
stop and submit a second market sell against shares that may no longer
exist, which is exactly the "opens a short on a long-only account"
failure cancel-before-sell exists to prevent.
"""
return [
holder
for holder in holders
if holder.entry_confidence is not None
and holder.strategy_id is not None
and holder.symbol.upper() not in sold_this_cycle
]
def _evict_discovered_holder(
holder: _DiscoveredHolder,
*,
new_symbol: str,
new_confidence: float,
margin: float,
executor,
trade_repository,
decision_repository,
outcome_repository,
decided_at: datetime,
is_paper: bool,
) -> SymbolOutcome:
"""Sell the weakest currently-held discovered position to free a slot.
Goes through `executor.exit_position` — the exact same cancel-before-sell
path a strategy's own SELL signal uses in `_process_symbol` — never a
bare broker call. Nothing about *how* this sells is special-cased for
eviction; only *why* it happened is: `reason` and `inputs_json.eviction`
name the new candidate and both confidences, so a later read of
`decisions` can tell an eviction sell from a signal-driven one without
guessing from prose.
Deliberately does **not** touch `discovered_cap.held` or fund
`new_symbol` in this same cycle. This sell has not filled yet — it is a
market order, not a completed exit — so this cycle's own headroom
arithmetic must keep seeing the account exactly as the top of the cycle
read it; only `positions_by_symbol`, re-read fresh from the broker at the
*start* of the *next* cycle, can honestly say the slot is empty. That
also means the freed slot is not reserved for `new_symbol` specifically:
next cycle's ranking runs in full again, and whichever candidate then
ranks best gets the slot — continuous top-N occupancy, not a one-for-one
swap. This mirrors how issue #86's own design doc reasoned about not
making two allocation decisions atomically in one cycle.
`veto` and `rejection_reason` are left `None` on the evicted symbol's own
row: nothing here overrode a strategy's signal (no signal was even read
for this call), so `action='sell'` reads exactly like a normal executed
sell — and, not incidentally, carries no `signal_action` key in its
`inputs_json`, so `simulation/step.py`'s `_parse_signal` reads it as
`(None, None)` and the shadow-portfolio simulator never replays this
real-money sell into its own state. The simulator computes its own,
entirely separate eviction locally instead (`_step_bar`'s own mirror),
per the same isolation rule that keeps `simulation/` from importing this
module at all.
"""
order = executor.exit_position(holder.position)
trade_id = (
None
if order is None
else trade_repository.record_submission(order, holder.strategy_id, is_paper)
)
reason = (
f"Evicted from its discovered slot to make room for {new_symbol}: "
f"{new_symbol}'s confidence ({new_confidence:.3f}) exceeds this "
f"holding's entry confidence ({holder.entry_confidence:.3f}) by more "
f"than the configured margin ({margin:.3f}). The freed slot becomes "
"available starting next cycle, once this sell is reconciled — not "
f"necessarily to {new_symbol}, which is still being rejected this "
"cycle."
)
return _record_decision(
decision_repository=decision_repository,
decided_at=decided_at,
strategy_id=holder.strategy_id, # type: ignore[arg-type]
symbol=holder.symbol,
action="sell",
reason=reason,
strategy_inputs={},
trade_id=trade_id,
order_id=None if order is None else order.order_id,
extra_inputs={
"eviction": {
"evicted_for_symbol": new_symbol,
"evicted_entry_confidence": holder.entry_confidence,
"new_candidate_confidence": new_confidence,
"eviction_margin_confidence": margin,
}
},
)
def _check_end_of_cycle_protection(*, broker) -> tuple[list[str], list[str], list[str]]:
"""What this cycle is handing back unprotected or still in flight — loudly.
Requirements §8: "The app must never terminate quietly while it believes
money is unprotected. Silence must mean 'protected', never 'not checked'."
On 2026-08-07 a daemon cycle submitted two buys, logged a clean summary,
and exited 0 while both sat unfilled at the broker — nothing checked for
that state, so nothing warned about it, and it took three days to notice
by hand. Tasks 1-4 make that specific failure much harder to reach (the
entry and its stop are now one atomic OTO request), but this check does
not trust that: it asks the broker directly, at the very end of every
cycle, after allocation and reconciliation have both already had their
turn — so it reports the state actually being handed to whichever process
(or none) looks next, from any cause, including a cause nobody has
thought of yet.
Two-sided, deliberately. "No protective SELL" is the obvious failure;
"more than one protective SELL" is the worse one, because two sell claims
on the same shares can both execute and the second opens a short on a
long-only account — which is why cancel-before-sell exists at all. Asking
only `is None` would have stayed silent through every scenario the
whole-branch review raised about a partially filled OTO parent. Both sides
are counted from the same broker-wide `open_orders` read this function
already makes for `outstanding_entries`, so the second side costs no extra
broker call — and counting from that list rather than calling the
executor's private `_protective_order_for` once per symbol also stops this
check reaching across a module boundary into a private method.
A broker read failing here is not evidence of "nothing to report" — that
is exactly the silent, not-checked failure this function exists to catch
— so a `TraderError` is logged loudly in its own right, and all three lists
come back empty only because nothing could be verified, never because
everything was confirmed safe.
"""
try:
positions = broker.get_positions()
open_orders = broker.get_open_orders()
except TraderError as exc:
_logger.error(
"could not verify end-of-cycle protection: %s; whether anything "
"is unprotected or still outstanding is UNKNOWN this cycle, not "
"clear",
exc,
)
return [], [], []
# Any open SELL is a claim on that symbol's shares, whatever its order
# type — the same widened definition `OrderExecutor._protective_order_for`
# uses, and for the same reason: a second claim is a second claim whether
# it is a trailing stop, a fixed stop, or a limit sell placed by hand.
sells_by_symbol: dict[str, list[SubmittedOrder]] = {}
for order in open_orders:
if order.side is OrderSide.SELL:
sells_by_symbol.setdefault(order.symbol, []).append(order)
unprotected: list[str] = []
double_protected: list[str] = []
for position in positions:
claims = sells_by_symbol.get(position.symbol, [])
if not claims:
unprotected.append(position.symbol)
_logger.warning(
"%s: cycle ending with a position and no protective order on "
"record — this exposure is UNPROTECTED",
position.symbol,
)
elif len(claims) > 1:
double_protected.append(position.symbol)
_logger.warning(
"%s: cycle ending with %d open protective SELL orders (%s) "
"against one position of %s shares — DOUBLE PROTECTED; if both "
"execute, the second sells shares this account does not hold, "
"which opens a short on a long-only account",
position.symbol,
len(claims),
", ".join(order.order_id for order in claims),
position.quantity,
)
outstanding_entries: list[str] = []
for order in open_orders:
if order.side is OrderSide.BUY:
outstanding_entries.append(order.symbol)
_logger.warning(
"%s: cycle ending with entry order %s still unfilled at the "
"broker; nothing will be watching it if this process exits "
"before the next cycle runs",
order.symbol,
order.order_id,
)
return unprotected, double_protected, outstanding_entries
def _cancel_entries_near_close(
*,
executor,
broker,
trade_repository,
pipeline_config: PipelineConfig,
decided_at: datetime,
) -> list[str]:
"""Cancel this app's own unfilled entries once the close is near.
The entry became a GTC OTO request to fix issue #3 (a DAY parent would
give its protective stop leg a DAY time-in-force too, leaving the
position naked overnight) — but GTC means nothing else expires an entry
that never fills. Left alone, a stale unfilled BUY holds exposure against
`max_total_exposure_pct` and blocks a fresh attempt via `pending_entry_for`
forever. This app has to own that expiry itself; this is where it does.
Deliberately **not** scoped to `run_once`'s own `symbols` — an earlier
version passed them through to `cancel_stale_entries`, and review caught
the hole that opened: an unfilled BUY creates no position, so a
**discovered** symbol whose entry never fills and whose news story then
drops out of the scan appears in neither `discovered` nor `held`, and
`fixed_tickers` never contains a pure-discovery symbol either. For that
population, scoping the sweep to `symbols` reintroduced exactly the
indefinitely-open GTC exposure this whole method exists to close.
`OrderExecutor.cancel_stale_entries` now sweeps every open order at the
broker directly, so this function no longer even has a symbol list to
pass it.
The clock is read here, not inside `cancel_stale_entries`: that method
knows how to sweep, and only `run_once` has `MarketClock.next_close` to
decide *when*. Outside the configured window, nothing is touched — cheap,
since `broker.get_clock()` is the only extra call made in that case.
A cancelled order's `trades` row is stamped `settled_at` **only when the
order shows no fill**, and that condition is the whole point. `settled_at`
is exactly what `reconcile_fills` uses to skip an order, so settling
unconditionally settles a *partially filled* entry before its fill was ever
reconciled, and those shares then never reach the ledger at all. Measured
A/B on a 20-of-29-share partial: left unsettled, the next cycle records a
round trip of ('20', '129.30'); settled, there is no round trip and no
trade — 20 real shares at the broker with nothing on record. So the test is
`_shares_moved`, the same one `reconcile_fills` uses to decide a fill
happened at all, read off the snapshot taken before the cancel. An order
that filled between that snapshot and the cancel therefore also stays
unsettled, which is the safe direction: an unsettled filled order is
reconciled next cycle, while a settled one is invisible forever. The cost
of leaving it unsettled is the issue #1 window, which is bounded and
already tracked; the cost of settling it is lost shares.
The order itself has to be captured *before* `cancel_stale_entries` runs:
once an order is cancelled it is no longer open, so it no longer appears in
`get_open_orders()` at all. The lookup here is a second broker-wide read,
not a per-symbol one — the same single-round-trip shape as the exposure
seed above, not the O(N) sweep a per-symbol lookup would be.
`mark_settled` no-ops for an id with no matching row (a dry-run has none),
so this never invents a row.
Never raises: a clock or open-orders read failure here must not cost the
cycle, the same discipline as every other guarded phase in `run_once`.
"""
try:
clock = broker.get_clock()
except TraderError as exc:
_logger.error(
"could not read the market clock: %s; not cancelling any stale "
"entries this cycle",
exc,
)
return []
threshold = timedelta(minutes=pipeline_config.cancel_entries_before_close_minutes)
if clock.next_close - decided_at > threshold:
return []
try:
pending_orders = {
order.symbol: order
for order in broker.get_open_orders()
if order.side is OrderSide.BUY
}
cancelled = executor.cancel_stale_entries()
except TraderError as exc:
_logger.error(
"could not sweep open orders for stale entries: %s; not "
"cancelling any this cycle",
exc,
)
return []
for symbol in cancelled:
order = pending_orders.get(symbol)
if order is None:
continue
if _shares_moved(order):
# Deliberately left unsettled: `reconcile_fills` skips a settled
# order, so stamping this one would discard shares that really
# moved. It is picked up next cycle instead.
_logger.warning(
"%s: cancelled entry %s had already filled %s share(s); "
"leaving its trades row unsettled so the fill is still "
"reconciled, rather than settling it and losing the shares",
symbol,
order.order_id,
order.filled_qty if order.filled_qty is not None else "some",
)
continue
trade_repository.mark_settled([order.order_id], decided_at)
return cancelled
def _record_protective_order(
order: SubmittedOrder,
*,
trade_repository,
outcome_repository,
is_paper: bool,
strategy_id: str | None = None,
) -> int | None:
"""Put a protective stop into `trades`. Returns the row id, or `None`.
`strategy_id` is taken from the open round trip the stop protects: the
stop belongs to the *position*, not to whichever strategy happened to
place it, and inventing an id would misattribute P/L. The `strategy_id`
argument is the fallback used only when there is no trip on record, and it
exists for one caller: an OTO's protective leg is recorded at *submission*
time, before the entry has filled, so no round trip can exist yet — but the
strategy that caused it is known right there and nowhere later. Left
unset (the reconcile and ratchet callers), no trip on record still means
`None`, because the round trip is the attribution of record either way.
Shared by `reconcile_protection` and `ratchet_protection`'s callers below:
both are idempotent in the sense that matters here (a stop returned by
either is always a *new* broker order id, never one already recorded), so
a stop from either one is always a new row. `ratchet_protection`'s
replacement order in particular is not optional bookkeeping — it is what
makes the eventual fill of a *ratcheted* stop recognisable at all, exactly
as for `reconcile_protection`'s stop. The `exists_for_order` check is a
guard against either of those idempotency claims ever breaking:
`trades.alpaca_order_id` is uniquely constrained, so a second write of the
same id would raise mid-cycle rather than being ignored.
"""
if trade_repository.exists_for_order(order.order_id):
return None
trip = outcome_repository.open_round_trip_for(order.symbol)
return trade_repository.record_submission(
order, trip.strategy_id if trip is not None else strategy_id, is_paper
)
def _record_decision(
*,
decision_repository,
decided_at: datetime,
strategy_id: str,
symbol: str,
action: str,
reason: str,
strategy_inputs: Mapping[str, object],
trade_id: int | None = None,
order_id: str | None = None,
extra_inputs: Mapping[str, object] | None = None,
rank: int | None = None,
veto: str | None = None,
rejection_reason: str | None = None,
outcome_note: str | None = None,
) -> SymbolOutcome:
"""Write one decision row and return its outcome.
A module-level function rather than a closure over `_process_symbol`'s
locals, because allocation happens in a different phase and must record
with the *candidate's own* snapshot rather than whatever the strategy
instance holds by then.
`veto` names the *class* of thing that overwrote the strategy's action, and
is absent when nothing did — silence means "nothing vetoed this", never "not
checked". It is machine-readable on purpose: the shadow portfolio has to
tell a veto it should ignore from one it should honour, and matching on the
prose in `reason` would break the moment a message is reworded.
`rejection_reason` (issue #21) is the specific gate within that class —
`price_floor` versus `discovery_cap`, both `screening`/`capacity`
respectively but not interchangeable for a "why didn't it buy X" query.
Written to its own `decisions` column, never folded into `inputs_json`,
so it survives independently of whether a row carries `inputs_json` at
all. `None` whenever `veto` is `None`: nothing overwrote the signal, so
there is nothing to name.
`outcome_note` (issue #25) is WHAT happened — an execution summary, kept
apart from `reason`/`reasoning` (WHY). `reason` must stay the strategy's
own rationale even for a decision that placed an order; it must never be
overwritten with execution prose the way it used to be, or "why" reads
"what" again.
"""
merged: dict[str, object] = dict(strategy_inputs)
if extra_inputs:
merged.update(extra_inputs)
if veto is not None:
merged["veto_class"] = veto
decision_repository.record(
decided_at=decided_at,
strategy_id=strategy_id,
ticker=symbol,
action=action,
reasoning=reason,
trade_id=trade_id,
inputs=merged or None,
rejection_reason=rejection_reason,
outcome_note=outcome_note,
)
return SymbolOutcome(
symbol, strategy_id, action, reason, trade_id, order_id, rank, rejection_reason
)
def _confidence_of(signal) -> float | None:
"""The strategy's own confidence, or `None` when it reports none.
`LlmStrategy` puts one in `Signal.indicators`; the rule strategies put
nothing there. `None` and `0.0` are different claims and must stay
distinguishable — a missing confidence rendered as zero would say the
strategy expressed minimum conviction, which it never did.
"""
raw = (signal.indicators or {}).get("confidence")
if isinstance(raw, bool) or raw is None:
return None
if isinstance(raw, (int, float, Decimal)):
value = float(raw)
return value if math.isfinite(value) else None
return None
def _confidence_sizing_factor(confidence: float | None, floor: Decimal) -> Decimal:
"""`floor + (1 - floor) * confidence` (issue #85, Decision 1).
A BUY decision already cleared the model's bar, so confidence modulates
conviction rather than gating the trade a second time — `floor` at
`confidence=0`, the unmodified `cap` at `confidence=1`, never a bare
`confidence` multiply (that would size a 0-confidence BUY at nothing,
too aggressive for a decision that already passed).
`None` (a rule strategy, which reports no confidence at all, per
`_confidence_of`'s own docstring) sizes at `floor` — the same
"missing is not the best case" reasoning as `_ABSENT_CONFIDENCE` in
`ranking/base.py`, not at the unmodified `cap`. Clamped to `[0, 1]`
defensively even though `parse_decision` already clamps the LLM path's
own confidence there (`llm/prompt.py`); this formula sizes real money
and must not trust an upstream guarantee it cannot itself verify.
`Decimal(str(...))`, never `Decimal(...)` directly, so a value like
`0.1` converts via its exact decimal string rather than its binary
float representation.
"""
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(strategy_inputs: Mapping[str, object]) -> Decimal | None:
"""Read `atr_pct_of_price` back out of a decision's own inputs snapshot.
`strategy_inputs["technicals"]` is exactly the dict `build_technical_
summary` returned (`llm/prompt.py`), captured for `decisions.inputs_json`
provenance — reused here, not recomputed, per issue #88's own scope
("reuse it, do not recompute"). Only `LlmStrategy` populates
`last_inputs`/`technicals` at all; a rule strategy's `strategy_inputs` is
`{}` (`_process_symbol`'s `dict(getattr(strategy, "last_inputs", {}) or
{})`), so this returns `None` for every rule-strategy entry, never a
fabricated reading.
`None` for anything unreadable — a missing `technicals` key, a
non-mapping value, a missing/`None` `atr_pct_of_price`, or a value that
does not parse as a `Decimal` (`build_technical_summary` always writes
either `None` or `str(Decimal)`, but this reads a persisted-shape
snapshot rather than trusting that upstream guarantee, the same
reasoning `_confidence_sizing_factor`'s defensive clamp documents) —
matching this codebase's "everything that can go wrong resolves to the
safe default" discipline rather than raising out of a sizing formula.
"""
technicals = strategy_inputs.get("technicals")
if not isinstance(technicals, Mapping):
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)` (issue #88, Decision 4).
Deliberately shrink-only: a calm name (`atr_pct_of_price <=
target_atr_pct`) sizes at the unmodified `cap`, never *above* it — this
is a risk-reduction lever, not a way to lever up a calm name. A volatile
name (`atr_pct_of_price > target_atr_pct`) sizes down proportionally.
Missing ATR data — `None`, or a non-positive reading, which would divide
by zero or invert the shrink direction — sizes at `1.0`, the unmodified
`cap`, deliberately unlike `_confidence_sizing_factor`'s choice to size
a missing `confidence` at `floor`. That choice is specific to
confidence: a `None` there means the *strategy itself* chose to report
no conviction, which is itself a fact worth sizing down for. A missing
ATR reading says nothing about the symbol at all — it means this
strategy never computed one (a rule strategy) or the warmup window
wasn't satisfied — so treating it as "maximally volatile" would punish
absence of data, not volatility. This mirrors the "missing is not the
worst case" discipline used for absent analyst coverage and absent
relative strength elsewhere in this codebase. A non-positive
`atr_pct_of_price` (ATR computed as exactly zero, which a dead-flat
window can legitimately produce) is mathematically the same answer:
`target / 0` is undefined, and the sizing question it would be
answering — "is this name calmer than the reference?" — is trivially
yes, so `1.0` (no shrink) is correct either way, not merely a fallback.
"""
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)
def _process_symbol(
*,
symbol: str,
run: StrategyRun,
account,
position,
halted: bool,
cache,
executor,
trade_repository,
decision_repository,
pipeline_config: PipelineConfig,
decided_at: datetime,
is_paper: bool,
sold_this_cycle: dict[str, str],
is_discovered: bool = False,
entry_blocked: Mapping[str, FilterOutcome] | None = None,
benchmark_bars: Sequence[Bar] | None = None,
benchmark_symbol: str | None = None,
) -> SymbolOutcome | _DeferredEntry:
"""Evaluate one symbol against one strategy — phase 1 of the cycle.
Never raises a `TraderError`: one bad symbol/strategy pair must abort its
own iteration and not the cycle. The failure is recorded as a Decision
with its reason, so the run is still fully accounted for.
Returns a `SymbolOutcome` for everything that is finished — a hold, a halt,
an entry screen, an executed SELL, an evaluator rejection, an error — and a
`_DeferredEntry` for an approved BUY, which cannot be finished here because
whether there is room for it depends on every *other* candidate this cycle.
Nothing in here consults `guardrails` or either counter; allocation owns all
three.
`sold_this_cycle` maps an upper-cased symbol to the strategy that has already
exited it in this cycle, and is **mutated in place** for the same reason
`ExposureCounter` is: a snapshot taken before the loop is blind to what the
loop itself does. Sells are submitted inline here, one (symbol, strategy)
pair at a time, so with more than one trading strategy this is the only thing
standing between two SELL signals on one position and two full-quantity
market sells — the second of which sells shares that no longer exist and
opens a short on a long-only account.
"""
strategy = run.strategy
# Read once into a typed local, exactly as `_fund` does: `StrategyRun.strategy`
# is deliberately typed `object`, so every dotted read of it is an untyped
# access and one suppression is enough.
strategy_id: str = strategy.id # type: ignore[attr-defined]
# What *this* symbol's own `evaluate()` call saw. Populated from
# `strategy.last_inputs` immediately after that call and never read off
# the strategy again, because `LlmStrategy.last_inputs` is per-instance
# mutable state and one instance serves every symbol in the cycle. Reading
# it at record time meant a symbol that failed *before* evaluating — no
# bars, or any `TraderError` — persisted the previous symbol's news, bar
# summary, `position_open` and `prompt_sha256`. That hash is a valid hash
# of the wrong symbol's prompt, so the acceptance check that rebuilds the
# prompt from `inputs_json` reports MATCH on a mislabelled row: the
# corruption is invisible to the one check meant to prove provenance, and
# `decisions.inputs_json` is the one artefact that cannot be rebuilt
# later. This local is scoped to one (symbol, strategy) pair, so a
# pre-evaluate path records nothing rather than something stale.
strategy_inputs: dict[str, object] = {}
def record(
action: str,
reason: str,
trade_id: int | None = None,
order_id: str | None = None,
extra_inputs: dict[str, object] | None = None,
veto: str | None = None,
rejection_reason: str | None = None,
) -> SymbolOutcome:
# Delegates to the module-level helper, which allocation also uses, so
# both phases write a `decisions` row exactly one way. `strategy_inputs`
# is read at call time on purpose: an LLM strategy exposes what it saw
# via `last_inputs`, populated below immediately after `evaluate()`; a
# rule strategy has no such attribute, so this stays `{}`.
return _record_decision(
decision_repository=decision_repository,
decided_at=decided_at,
strategy_id=strategy_id,
symbol=symbol,
action=action,
reason=reason,
strategy_inputs=strategy_inputs,
trade_id=trade_id,
order_id=order_id,
extra_inputs=extra_inputs,
veto=veto,
rejection_reason=rejection_reason,
)
try:
start = decided_at - timedelta(days=pipeline_config.evaluation_lookback_days)
bars = cache.ensure(symbol, start, decided_at)
if not bars:
return record("error", f"No bars available for {symbol}.")
# Duck-typed, not a `Strategy` Protocol method: only `LlmStrategy`
# implements this (issue #87), the mirror image of `last_inputs`
# being read via `getattr` below rather than added to the Protocol.
# A strategy that doesn't have it (the three non-LLM strategies)
# simply never gets the call.
set_benchmark_bars = getattr(strategy, "set_benchmark_bars", None)
if callable(set_benchmark_bars):
set_benchmark_bars(benchmark_bars, benchmark_symbol)
signal = strategy.evaluate(bars, position)
strategy_inputs = dict(getattr(strategy, "last_inputs", {}) or {})
confidence = _confidence_of(signal)
# What the strategy *said*, before any veto below can overwrite
# `action`. Captured here — the one point every path passes through
# after `evaluate()` — so it reaches every record site uniformly,
# including allocation's snapshot and the same-symbol collision loser.
#
# Without it, `decisions.action` answers "what did the pipeline do",
# and only a strategy that cannot trade has an `action` that is still
# its own opinion: `is_live` returns above, before every veto. So the
# one strategy in `mode: trading` — the control the shadow portfolio
# calibrates against — was the only contaminated series, and a *rule*
# strategy promoted to trading would have had no recoverable signal at
# all, since it populates no `last_inputs`. Measured 2026-08-11:
# `turtle_20_10`'s `inputs_json` keys were `[]`, and one `ollama_news`
# row read `action='rejected'` over `raw_response.action='buy'`.
strategy_inputs["signal_action"] = signal.action.value
# Always written, even as null: `None` (expressed no view) and `0.0`
# (expressed minimum conviction) are different claims, and a key that
# vanishes when absent makes them indistinguishable from a reader's
# point of view.
strategy_inputs["signal_confidence"] = confidence
last_close = bars[-1].close
# A shadow strategy stops here: decision recorded, nothing submitted,
# and — because this return precedes every check below — no evaluator
# call either. The evaluator exists to gate trading, and a 400-bar
# backtest (or worse, an LLM strategy's model calls) every cycle is
# not a price a strategy that cannot trade should pay.
if not run.is_live:
return record(signal.action.value, signal.reason)
if position is not None:
if signal.action is Action.SELL:
claimed_by = sold_this_cycle.get(symbol.upper())
if claimed_by is not None:
# `capacity`, not `screening`: nothing is wrong with this
# signal or this symbol. Another strategy already claimed
# these shares, and there is only one position to exit.
return record(
"rejected",
f"{symbol} was already sold this cycle by {claimed_by}, "
f"which is the exit being acted on. There is one position "
f"per symbol, so a second sell would offer shares that no "
f"longer exist and open a short. Signal was: "
f"{signal.reason}",
veto=_VETO_CAPACITY,
rejection_reason=_REJECTION_ALREADY_EXITED,
)
# Claimed *before* submitting, mirroring `ExposureCounter` being
# advanced before an entry goes out. If `exit_position` raises,
# the claim still stands and the next strategy is refused rather
# than allowed to retry the sell: refusing costs at most one
# cycle's exit, while a double sell opens a short. This app never
# retries an order for the same reason.
sold_this_cycle[symbol.upper()] = strategy_id
order = executor.exit_position(position)
trade_id = (
None
if order is None
else trade_repository.record_submission(order, strategy_id, is_paper)
)
return record(
"sell",
signal.reason,
trade_id,
None if order is None else order.order_id,
)
# A BUY against an open position is ignored: no pyramiding in this
# slice, so an existing long means there is nothing to do.
return record(
"hold",
f"Already long {symbol}. {signal.reason}",
veto=_VETO_CAPACITY,
rejection_reason=_REJECTION_ALREADY_LONG,
)
if signal.action is not Action.BUY:
return record("hold", signal.reason)
# Screened out for purchase, but deliberately still evaluated above:
# the operator's fixed tickers are a watchlist, not a buy list, so a
# failing filter blocks the entry and leaves the decision on record.
# Placed after the SELL branch on purpose — a holding that later fails
# a filter must still be sellable by a strategy, or the broker's
# trailing stop becomes its only exit.
blocked = (entry_blocked or {}).get(symbol.upper())
if blocked is not None:
# `screening`, not `capacity`: untradable, a warrant, too cheap or
# too thin are facts about the *symbol*, equally true of a simulated
# portfolio, so the shadow portfolio honours this one. `blocked.code`
# is one of `discovery.filters.REASON_*` — whichever gate in the
# chain actually rejected this symbol, never re-derived from
# `blocked.reason`'s prose.
return record(
"rejected",
blocked.reason,
veto=_VETO_SCREENING,
rejection_reason=blocked.code,
)
# An unfilled entry order is not a position, but it is exposure already
# committed. Without this the next cycle buys the same symbol again.
# Found by running the cycle twice against the paper account.
pending = executor.pending_entry_for(symbol)
if pending is not None:
return record(
"hold",
f"An unfilled entry order for {symbol} is already open "
f"({pending.quantity} at {pending.limit_price}, order "
f"{pending.order_id}). Not placing a second one.",
veto=_VETO_CAPACITY,
rejection_reason=_REJECTION_PENDING_ENTRY,
)
if halted:
return record(
"halted",
f"Daily loss limit reached, so the BUY on {symbol} was not acted "
f"on. Signal was: {signal.reason}",
veto=_VETO_CAPACITY,
rejection_reason=REASON_DAILY_LOSS_HALT,
)
if run.evaluator is not None and strategy.backtestable:
verdict = run.evaluator.evaluate(symbol, bars, signal)
if not verdict.approved:
# `screening`: the evaluator judges the *signal's* historical
# quality on this symbol, which a simulated portfolio's own cash
# does nothing to change.
return record(
"rejected",
verdict.reason,
extra_inputs=verdict.evidence,
veto=_VETO_SCREENING,
rejection_reason=_REJECTION_EVALUATOR,
)
evaluator_reason = verdict.reason
evidence = verdict.evidence
else:
# A strategy that cannot be backtested cannot be gated on one: a
# 400-bar window would be 400 model calls per symbol per cycle
# just to decide whether to let the signal through.
evaluator_reason = "evaluator skipped: strategy is not backtestable"
evidence = {"evaluator": "skipped: strategy is not backtestable"}
# Approved. What is left is whether there is room, which cannot be
# decided one symbol at a time — the whole point of the phase split.
# `confidence` was read once immediately after `evaluate()`, so the
# value the ranker sizes on and the value recorded on the decision row
# cannot disagree.
return _DeferredEntry(
symbol=symbol,
run=run,
signal=signal,
last_close=last_close,
evaluator_reason=evaluator_reason,
evidence=dict(evidence),
strategy_inputs=dict(strategy_inputs),
is_discovered=is_discovered,
candidate=Candidate(
symbol=symbol,
confidence=confidence,
reason=signal.reason,
last_close=last_close,
comparables=build_comparables(bars, strategy.warmup_bars()),
),
)
except TraderError as exc:
_logger.error("%s failed: %s", symbol, exc)
return record("error", str(exc))
def _rank_candidates(
candidates: Sequence[_DeferredEntry],
ranker: Ranker | None,
confidence_ordering_enabled: bool = False,
) -> Ranking:
"""The ranked order, and how it was arrived at. Never raises.
The second of two layers of containment. `LlmRanker.rank` already catches
`TraderError`; this catches everything else, so a bug in a ranker
implementation degrades the *order* rather than costing the cycle. Ranking
must never be able to stop trading — the same rule discovery follows, and
it is safe here for a specific reason: the ranker only orders, so a
degraded order cannot unlock risk that the guardrails would otherwise have
refused.
`confidence_ordering_enabled` (issue #86) short-circuits straight to the
confidence-descending order, *whether or not* `ranker` is usable — the
ranker is not even called. Checked first, ahead of the `ranker is None`
branch below: without this ordering, a configured `LlmRanker` would still
run (and be billed a model call) only to have its order discarded, which
both wastes the call and makes `ranker.calls` lie about whether ranking
happened this cycle.
"""
views = [entry.candidate for entry in candidates]
if confidence_ordering_enabled:
base = fallback_ranking(views)
return Ranking(
order=base.order,
source=SOURCE_CONFIDENCE_ORDERING_ENABLED,
inputs={"rank_confidence_ordering_enabled": True},
)
if ranker is None:
return fallback_ranking(views)
try:
ranking = ranker.rank(views)
except Exception as exc: # noqa: BLE001 - ranking must never cost a cycle
_logger.error("ranking failed, falling back to confidence order: %s", exc)
base = fallback_ranking(views)
return Ranking(
order=base.order,
source=base.source,
inputs={"rank_error": str(exc)},
)
# A ranker that returns something other than a permutation of what it was
# given would silently drop a candidate — vetoing, which is not its job.
if sorted(ranking.order) != sorted(c.symbol for c in views):
_logger.error(
"ranker returned %s for candidates %s; falling back to confidence "
"order rather than dropping or inventing a candidate",
list(ranking.order),
[c.symbol for c in views],
)
base = fallback_ranking(views)
return Ranking(
order=base.order,
source=base.source,
inputs={"rank_error": "ranker did not return a permutation"},
)
return ranking
def _one_candidate_per_symbol(
candidates: Sequence[_DeferredEntry],
) -> tuple[list[_DeferredEntry], list[tuple[_DeferredEntry, _DeferredEntry]]]:
"""Split the pool into one entry per symbol, plus the collisions.
Returns `(unique, collided)`, where `collided` pairs each losing entry with
the one that beat it. The winner is the **first** entry for that symbol,
which is deterministic: phase 1 walks `symbols` in universe order and
`strategies` in configured order, so the same inputs always elect the same
candidate and record the same rows.
Why this exists at all, since `StrategiesConfig._at_most_one_trading`
already rejects a second `mode: trading` and a shadow strategy never becomes
a candidate: **this is not dead code, do not delete it.** `run_once` is a
public, fully-injected function whose `strategies` argument is a plain
sequence, and every other invariant in it is defended where it is relied on
rather than assumed from config. Without this, two trading strategies
approving one symbol collide in `_allocate`'s symbol-keyed dict — measured
as six limit buys for three symbols, each submitted twice, all six rows
attributed to the last strategy — and `pending_entry_for` cannot catch it
because both submissions land inside the same cycle. Deduplicating *before*
ranking rather than after is deliberate: two `Candidate`s sharing a symbol
would also collide in `parse_ranking`'s own symbol lookup.
"""
unique: list[_DeferredEntry] = []
winners: dict[str, _DeferredEntry] = {}
collided: list[tuple[_DeferredEntry, _DeferredEntry]] = []
for entry in candidates:
winner = winners.get(entry.symbol)
if winner is None:
winners[entry.symbol] = entry
unique.append(entry)
else:
collided.append((entry, winner))
return unique, collided
def _allocate(
*,
candidates: Sequence[_DeferredEntry],
ranker: Ranker | None,
account,
guardrails,
executor,
trade_repository,
decision_repository,
outcome_repository,
pipeline_config: PipelineConfig,
decided_at: datetime,
is_paper: bool,
discovered_cap: _DiscoveredCapCounter,
max_discovered_positions: int,
exposure: ExposureCounter,
discovered_holders: list[_DiscoveredHolder] | None = None,
sold_this_cycle: dict[str, str] | None = None,
) -> list[SymbolOutcome]:
"""Fund the ranked candidates top-down until the headroom runs out.
`ExposureCounter` and `_DiscoveredCapCounter` stay mutable and keep
advancing inside this loop, exactly as they did when it lived inside the
symbol loop: a total computed once and never updated is blind to what the
cycle itself is spending, which is the documented bug that put twelve limit
buys on the account in one cycle. The only thing that changed is the order
they are fed in.
At most one entry per symbol is funded, whatever the strategy pool looks
like — this app holds one position per symbol and never pyramids. A
same-symbol collision is *recorded* rather than dropped: an approved
candidate that produced no `decisions` row would be a hole in the one table
that answers "why did nothing happen".
`discovered_holders` (issue #83), when eviction is enabled, is also
mutable across this loop: `_fund` removes a holder from it the moment
that holder is evicted, so a *second* capped-out candidate later in this
same ranking compares against whichever holder is now weakest among what
remains, never against one already sold two candidates ago. Defaults to
`None`/empty for every caller that predates issue #83 (and for every
cycle with `eviction_enabled=False`), which is exactly "nothing eligible
to evict" — the same behaviour as before this parameter existed.
"""
if not candidates:
return []
holders = discovered_holders if discovered_holders is not None else []
sold = sold_this_cycle if sold_this_cycle is not None else {}
pool, collided = _one_candidate_per_symbol(candidates)
outcomes: list[SymbolOutcome] = []
for loser, winner in collided:
_logger.warning(
"%s was approved by both %s and %s in one cycle; funding %s only, "
"because this app holds one position per symbol and never pyramids",
loser.symbol,
winner.run.strategy.id,
loser.run.strategy.id,
winner.run.strategy.id,
)
# Recorded with the loser's *own* id and its own `strategy_inputs`
# snapshot, never the winner's, and with no rank: it never entered the
# ranking, so claiming a place in it would be a false provenance claim.
outcomes.append(
_record_decision(
decision_repository=decision_repository,
decided_at=decided_at,
strategy_id=loser.run.strategy.id,
symbol=loser.symbol,
action="rejected",
reason=(
f"{loser.symbol} was already approved this cycle by "
f"{winner.run.strategy.id}, which is the candidate being "
f"funded. This app holds one position per symbol and never "
f"pyramids, so this second approval was not acted on. "
f"Signal was: {loser.candidate.reason}"
),
strategy_inputs=loser.strategy_inputs,
extra_inputs=loser.evidence,
# `capacity`: nothing is wrong with this signal or this symbol.
# It lost to a claim on the same shares, which is a fact about
# *this* account holding one position per symbol — a simulated
# portfolio applies its own version of that rule.
veto=_VETO_CAPACITY,
rejection_reason=_REJECTION_DUPLICATE_APPROVAL,
)
)
ranking = _rank_candidates(
pool,
ranker,
confidence_ordering_enabled=pipeline_config.confidence_ordering_enabled,
)
by_symbol = {entry.symbol: entry for entry in pool}
pool_size = len(pool)
_logger.info(
"ranked %d entry candidate(s) by %s: %s",
pool_size,
ranking.source,
", ".join(ranking.order),
)
for position_index, symbol in enumerate(ranking.order, start=1):
entry = by_symbol[symbol]
provenance: dict[str, object] = {
"rank_position": position_index,
"rank_pool_size": pool_size,
"rank_source": ranking.source,
"rank_order": list(ranking.order),
"rank_comparables": entry.candidate.comparables.as_dict(),
**dict(ranking.inputs),
}
note = ranking.notes.get(symbol)
if note:
provenance["rank_reason"] = note
outcomes.extend(
_fund(
entry=entry,
rank=position_index,
provenance=provenance,
account=account,
guardrails=guardrails,
executor=executor,
trade_repository=trade_repository,
decision_repository=decision_repository,
outcome_repository=outcome_repository,
pipeline_config=pipeline_config,
decided_at=decided_at,
is_paper=is_paper,
discovered_cap=discovered_cap,
max_discovered_positions=max_discovered_positions,
exposure=exposure,
discovered_holders=holders,
sold_this_cycle=sold,
)
)
return outcomes
def _fund(
*,
entry: _DeferredEntry,
rank: int,
provenance: dict[str, object],
account,
guardrails,
executor,
trade_repository,
decision_repository,
outcome_repository,
pipeline_config: PipelineConfig,
decided_at: datetime,
is_paper: bool,
discovered_cap: _DiscoveredCapCounter,
max_discovered_positions: int,
exposure: ExposureCounter,
discovered_holders: list[_DiscoveredHolder] | None = None,
sold_this_cycle: dict[str, str] | None = None,
) -> list[SymbolOutcome]:
"""Size, submit and record one ranked candidate. Returns one outcome, or
two when funding it triggers an eviction (issue #83): the candidate's own
rejection, and the evicted holder's sell.
Never raises a `TraderError`: one candidate failing on a broker or data
error must not cost the rest of the ranking, and a candidate that reaches
this function and produces no `decisions` row would be a silent hole in the
one table that answers "why did nothing happen".
Anything broader **does** propagate, out of `_allocate` and out of
`run_once`. That is the containment scope this code had before the phase
split and it is deliberately unchanged here, but it is worth naming rather
than papering over with "never raises". Measured: with `record_submission`
raising `OSError`, `run_once` propagates it and the limit buy is already at
the broker — one order placed, no `decisions` row, no `trades` row. Per
`reconcile_fills`, a fill with no `trades` row behind it is skipped as "not
ours", so that position's round trip is never attributed; it does still get
a protective stop, because `reconcile_protection` works from the broker's
positions rather than from this app's records. Widening the catch here would
not prevent any of that — the order is already placed by the time it could
fire, and this app never retries one.
"""
# Read once and reused, rather than `strategy.id` at each of the four
# places that need it: `StrategyRun.strategy` is deliberately typed
# `object`, so every dotted read of it is an untyped access, and one is
# enough.
strategy_id: str = entry.run.strategy.id # type: ignore[attr-defined]
symbol = entry.symbol
def record(
action: str,
reason: str,
trade_id=None,
order_id=None,
veto=None,
rejection_reason=None,
extra=None,
outcome_note=None,
):
return _record_decision(
decision_repository=decision_repository,
decided_at=decided_at,
strategy_id=strategy_id,
symbol=symbol,
action=action,
reason=reason,
strategy_inputs=entry.strategy_inputs,
trade_id=trade_id,
order_id=order_id,
extra_inputs={**entry.evidence, **provenance, **(extra or {})},
rank=rank,
veto=veto,
rejection_reason=rejection_reason,
outcome_note=outcome_note,
)
try:
limit_price = executor.limit_price_for(entry.last_close)
# Propose the largest position the cap allows, then let the guardrail
# be the single authority on the final number. Proposing and checking
# in one place would put sizing logic in two.
cap = account.portfolio_value * pipeline_config.max_position_pct / _HUNDRED
if pipeline_config.confidence_sizing_enabled:
cap = cap * _confidence_sizing_factor(
entry.candidate.confidence, 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 — two
# independent factors on the same `cap`, not a second gate.
cap = cap * _volatility_sizing_factor(
_atr_pct_of_price(entry.strategy_inputs),
pipeline_config.volatility_sizing_target_atr_pct,
)
proposed = max(int(cap // limit_price), 1)
risk = guardrails.check_entry(
account,
symbol,
proposed,
limit_price,
is_discovered=entry.is_discovered,
discovered_positions_held=discovered_cap.held,
max_discovered_positions=max_discovered_positions,
exposure=exposure,
)
if not risk.allowed:
# `capacity`: every reason `check_entry` refuses — no headroom under
# either cap, the discovered-position limit — is a fact about this
# account's money, not about the symbol or the signal. A simulated
# portfolio has its own cash and its own `ExposureCounter`, so it
# ignores this and applies its own. `risk.code` is one of
# `risk.guardrails.REASON_*`, never re-derived from `risk.reason`.
#
# Issue #22: only the discovered-position-cap veto (identified by
# `risk.code == REASON_DISCOVERY_CAP` — never the price-floor or
# total-exposure ones, which also land here) gets an
# `opportunity_cost` payload — the would-be entry price and the
# timestamp of the veto — recorded into this row's `inputs_json`,
# so a later, separate job can reconstruct what would have
# happened had the cap not fired. `limit_price` and `decided_at`
# are exactly what a real entry would have used and been recorded
# at, had `check_entry` allowed it. Nothing else about this cycle
# changes: no order goes out, no capital is committed.
opportunity_cost = None
eviction_outcome: SymbolOutcome | None = None
extra_fields: dict[str, object] = {}
if risk.code == REASON_DISCOVERY_CAP:
opportunity_cost = {
"opportunity_cost": {
"symbol": symbol,
"would_be_entry_price": limit_price,
"would_be_entry_price_at": decided_at.isoformat(),
"max_discovered_positions": max_discovered_positions,
"discovered_positions_held": discovered_cap.held,
}
}
extra_fields.update(opportunity_cost)
# Issue #83, gated behind `eviction_enabled` (`False` by
# default): a candidate capped out *only* by the
# discovered-position limit — every other guardrail already
# passed — may still evict the weakest current holder rather
# than being rejected outright. See `_evict_discovered_holder`
# for why this never funds `symbol` in this same cycle.
if (
pipeline_config.eviction_enabled
and entry.candidate.confidence is not None
):
eligible = _eligible_eviction_targets(
discovered_holders or [], sold_this_cycle or {}
)
if eligible:
weakest = min(
eligible,
key=lambda h: h.entry_confidence, # type: ignore[arg-type,return-value]
)
margin = pipeline_config.eviction_margin_confidence
# Strictly greater-than: a tie must never evict, or the
# margin is not a margin.
if (
entry.candidate.confidence > weakest.entry_confidence + margin # type: ignore[operator]
):
eviction_outcome = _evict_discovered_holder(
weakest,
new_symbol=symbol,
new_confidence=entry.candidate.confidence,
margin=margin,
executor=executor,
trade_repository=trade_repository,
decision_repository=decision_repository,
outcome_repository=outcome_repository,
decided_at=decided_at,
is_paper=is_paper,
)
# Mutate both shared, cycle-scoped structures so a
# *later* candidate in this same ranking neither
# re-evicts this holder nor lets `_process_symbol`
# (already finished for this cycle, but the dict
# is read defensively) treat it as sellable again.
if discovered_holders is not None:
discovered_holders.remove(weakest)
if sold_this_cycle is not None:
sold_this_cycle[weakest.symbol.upper()] = (
weakest.strategy_id # type: ignore[assignment]
)
extra_fields["eviction"] = {
"evicted_symbol": weakest.symbol,
"evicted_entry_confidence": weakest.entry_confidence,
"new_candidate_confidence": entry.candidate.confidence,
"eviction_margin_confidence": margin,
}
rejection = record(
"rejected",
risk.reason,
veto=_VETO_CAPACITY,
rejection_reason=risk.code,
extra=extra_fields or None,
)
return [rejection, eviction_outcome] if eviction_outcome else [rejection]
if entry.is_discovered:
# This entry is about to go out — real order or dry run, either
# way it is this cycle's own exposure, not the account's from
# before the cycle started. Incrementing here, before submission,
# is what makes the *next* ranked discovered candidate in this same
# loop see the higher count instead of the pre-loop snapshot.
discovered_cap.held += 1
# Advance the running total for the same reason the discovered count
# advances: the next candidate in this loop must see the headroom this
# entry just consumed, not the pre-loop snapshot.
exposure.commit(Decimal(risk.approved_quantity) * limit_price)
placed = executor.enter(symbol, risk.approved_quantity, entry.last_close)
if placed is None:
trade_id = trade_repository.record_dry_run(
symbol,
OrderSide.BUY,
risk.approved_quantity,
limit_price,
strategy_id,
is_paper,
)
return [
record(
"buy",
entry.signal.reason, # type: ignore[attr-defined]
trade_id,
outcome_note=f"DRY RUN: would buy {risk.approved_quantity} at "
f"{limit_price}. {entry.evaluator_reason}",
)
]
order = placed.entry
trade_id = trade_repository.record_submission(order, strategy_id, is_paper)
# The OTO's protective leg is a second broker order created by the same
# request, and it has to be recorded exactly like the entry — for the
# mirror-image reason. `reconcile_fills` skips any fill with no `trades`
# row as "not ours", so an unrecorded leg makes the *exit* it eventually
# performs invisible: measured, the leg fired 10.4% below a 129.32
# entry, the round trip stayed open, `total_realized_pl()` reported 0
# against a truth of -802.34, and the symbol then wedged forever on "a
# buy fill arrived with a round trip already open". `strategy_id` is
# passed explicitly because no round trip exists yet at submission
# time — the entry has not filled — and the strategy that caused this
# leg is known here and nowhere later.
if placed.protective_leg is not None:
_record_protective_order(
placed.protective_leg,
trade_repository=trade_repository,
outcome_repository=outcome_repository,
is_paper=is_paper,
strategy_id=strategy_id,
)
else:
_logger.error(
"%s: entry %s went out with no protective leg reported by the "
"broker; if it fills, this position is UNPROTECTED until "
"reconciliation repairs it",
symbol,
order.order_id,
)
return [
record(
"buy",
entry.signal.reason, # type: ignore[attr-defined]
trade_id,
order.order_id,
outcome_note=f"Bought {risk.approved_quantity} at limit {limit_price}. "
f"{entry.evaluator_reason}",
)
]
except TraderError as exc:
_logger.error("%s failed during allocation: %s", symbol, exc)
return [record("error", str(exc))]
def _fill_quantity(order: SubmittedOrder) -> Decimal:
"""The quantity that actually filled, falling back to what was requested.
A partial fill makes these two different; `filled_qty` is `None` only when
the broker has not reported it, not when the fill was for zero shares.
"""
return order.filled_qty if order.filled_qty is not None else order.quantity
def _shares_moved(order: SubmittedOrder) -> bool:
"""Whether this order actually moved shares, whatever its status says.
`AlpacaBroker.get_filled_orders` deliberately queries `CLOSED`, which
covers filled, cancelled and expired, and leaves it to the caller to
decide which matter. Deciding on `status == "filled"` alone discarded a
*partially filled and then expired* DAY entry — 20 of 29 shares, a real
position on the broker's books, no round trip opened, `trades.filled_qty`
never written, and the whole position's realized P/L lost when it was
eventually sold. The design names that case verbatim.
So the test is whether shares moved. When the broker reports no
`filled_qty` at all, a status of `"filled"` is taken at face value; any
other status is treated as having moved nothing, rather than falling back
to the *requested* quantity the way `_fill_quantity` does — that fallback
is right for sizing a known fill and would be wrong here, counting an
expired unfilled order as a fill.
"""
if order.filled_qty is not None:
return order.filled_qty > 0
return order.status == "filled"
def _fill_time(order: SubmittedOrder, now: datetime) -> datetime:
"""When the fill actually happened, preferring the broker's own report.
Falling back to `now` only when the broker hasn't reported a fill time
keeps `Trade.filled_at` stable across repeated cycles instead of drifting
forward every time an already-filled order is reconciled again.
"""
return order.filled_at if order.filled_at is not None else now
def _fill_price_for_ledger(order: SubmittedOrder) -> Decimal | None:
"""The price to write onto the trade's own ledger row.
`Trade.filled_avg_price` is nullable, so when neither the broker's real
fill price nor the order's limit price is known, this reports the gap as
`None` rather than fabricating a `0` — a `0` in a money column reads back
as a real, if worthless, fill, which is worse than an honest absence.
"""
if order.filled_avg_price is not None:
return order.filled_avg_price
if order.limit_price is not None:
return order.limit_price
_logger.warning(
"%s: order %s has no reported fill price or limit price; recording "
"no price on its trade row rather than a fabricated 0",
order.symbol,
order.order_id,
)
return None
def _fill_price_for_round_trip(
order: SubmittedOrder, position_price: Decimal | None
) -> Decimal:
"""The price to write onto a round trip's `entry_price`/`exit_price`.
Unlike the ledger, these columns are NOT NULL, so this must return a
number. `position_price` is the best number this app still has about the
position — `avg_entry_price` for an entry, the last known `current_price`
for an exit — and is used, with a loud warning, only when the order
itself carries neither a real fill price nor a limit price. Falling back
to `0` happens only when even that is unavailable, which should not occur
for a genuinely filled order; it is logged as a correctness risk rather
than treated as routine, per the same reasoning as the ledger's `None`.
"""
if order.filled_avg_price is not None:
return order.filled_avg_price
if order.limit_price is not None:
return order.limit_price
if position_price is not None:
_logger.warning(
"%s: order %s has no reported fill price or limit price; using "
"the position's last known price (%s) for its round trip",
order.symbol,
order.order_id,
position_price,
)
return position_price
_logger.warning(
"%s: order %s has no reported fill price, limit price, or known "
"position price; recording 0 as a last resort — this round trip's "
"realized P/L cannot be trusted",
order.symbol,
order.order_id,
)
return Decimal(0)
def _weighted_price(
orders: list[SubmittedOrder], position_price: Decimal | None
) -> Decimal:
"""Quantity-weighted average fill price across every order on one side
of a round trip.
Design: "entry_price and exit_price are quantity-weighted averages across
however many fills each side took." A plain average would let a small,
badly-priced fill distort the realized price as much as the large fill
that did most of the work — a two-order 5-and-5 split is a coincidence of
this being the reproduction case, not the general rule. Each order's own
price is resolved via `_fill_price_for_round_trip`, so a single order
missing a price falls back exactly as it would alone.
"""
total_qty = Decimal(0)
total_notional = Decimal(0)
for order in orders:
qty = _fill_quantity(order)
total_qty += qty
total_notional += qty * _fill_price_for_round_trip(order, position_price)
if total_qty == 0:
return Decimal(0)
return total_notional / total_qty
def _capped_weighted_exit(
orders: list[SubmittedOrder], cap: Decimal, position_price: Decimal | None
) -> tuple[Decimal, Decimal]:
"""Quantity and quantity-weighted price for an exit, never exceeding `cap`.
`cap` is the open trip's own recorded `quantity` — this app never sells
more than it holds on paper, so an exit's fills summing to more than that
(a broker-side quantity this app's own entry accounting never saw, e.g.
the multi-cycle split-entry gap this module documents) must not inflate
the recorded `quantity` or leak into the recorded price. Only the fills
(partially, for whichever fill crosses the cap) that make up the capped
quantity are weighted; excess beyond the cap is dropped from both the
quantity and the price, not merely from the quantity.
"""
remaining = cap
total_qty = Decimal(0)
total_notional = Decimal(0)
for order in orders:
if remaining <= 0:
break
qty = min(_fill_quantity(order), remaining)
total_notional += qty * _fill_price_for_round_trip(order, position_price)
total_qty += qty
remaining -= qty
if total_qty == 0:
return Decimal(0), Decimal(0)
return total_qty, total_notional / total_qty
def _fill_poll_since(trade_repository, now: datetime) -> datetime:
"""The lower bound for this cycle's `get_filled_orders` query.
Anchored on the oldest *outstanding submission*, never on the newest fill.
Alpaca's `after` filter matches `submitted_at`, so bounding on fill time
would exclude an order submitted days ago that fills today — which is what
a GTC trailing stop is, and stop-driven exits are this app's dominant exit
path.
The overlap absorbs clock skew between the timestamp this app recorded at
submission and the one Alpaca filters on. An hour is far more than the
observed difference and still bounds the query to a small window in the
steady state, where nothing is outstanding at all.
This used to carry a caveat reading "an order that was never filled and
never cancelled — an expired DAY limit buy, say — stays unsettled and keeps
the window open at its submission time". That was issue #1, and it grew: 12
such rows by 2026-08-13, the oldest 13 days stale. `settle_terminal_orders`
closes it, so the caveat is gone rather than merely softened — but the
reason it was true still is. `reconcile_fills` learns only about fills, so
**something other than this function has to settle a row whose order ended
without one.** If that sweep is ever removed, restore the caveat with it.
"""
oldest = trade_repository.oldest_unsettled_submitted_at()
return (oldest or now) - _FILL_POLL_OVERLAP
[docs]
def settle_terminal_orders(*, broker, trade_repository, now: datetime) -> list[str]:
"""Settle rows whose broker order ended without ever filling (issue #1).
`reconcile_fills` only ever learns about orders that *filled*: it stamps
`settled_at` when a fill is folded into round-trip state, and nothing else
does. An order that expires, is cancelled, or is rejected therefore stays
unsettled forever and holds `_fill_poll_since` open at its own submission
time. Measured on the live database 2026-08-13: 12 unsettled rows out of 44,
the oldest a `pending_new` SPY buy from 2026-07-31 17:52:35 — 13 days stale,
up from 4 such rows nine days earlier, so it grows at roughly one a day.
Alpaca answers `get_filled_orders` with at most 200 orders, and when that cap
is reached the *oldest* fills drop out silently. A stop-driven exit that goes
unreconciled is this app's dominant exit path going unattributed.
**The naive fix is dangerous, and this is not it.** Absence from
`get_open_orders` does not mean an order never filled — it usually means it
did. Settling on absence would stamp `settled_at` on a filled order,
`reconcile_fills` would then skip it forever, and those shares would never
reach the ledger at all: the exact failure the window exists to prevent, and
the one measured in `_cancel_entries_near_close`'s A/B on a 20-of-29 partial.
So this settles on **positive evidence of a terminal, non-filling outcome**
and on nothing else. Two conditions, both required:
1. the broker reports a status in `_TERMINAL_NON_FILLING_STATUSES` — asked
per order id, which is also the only query Alpaca's 200-order cap cannot
truncate; and
2. it reports `filled_qty` and that quantity is **zero**.
Condition 2 is `filled_qty == 0`, not `not _shares_moved(order)`, and the
difference is deliberate. `_shares_moved` answers "did shares move?" and
treats an absent `filled_qty` on a non-`filled` order as "no" — which is the
right reading when deciding whether to record a fill, because guessing "yes"
would invent one. Here the consequence is inverted: a wrong "no" settles the
row and hides real shares permanently. So an absent `filled_qty` is
*unknown*, not zero, and unknown does not settle. A partially filled
cancelled order is likewise left alone; `reconcile_fills` picks its fill up
and `mark_settled` stamps the row once the round trip commits, which is the
only path that may ever settle a row that moved shares.
The open-orders read is a cost filter, never evidence. An id in that list is
demonstrably still live, so there is nothing to settle and no reason to spend
a lookup on it — and in the steady state that is most of the population,
since every held position carries an open protective SELL whose row stays
legitimately unsettled until it fires. Skipping cannot cause a wrong settle,
because the settle decision is made from the per-id answer alone.
Never raises, the same discipline as every other guarded phase in
`run_once`: a failed read yields fewer settles, not a lost cycle. Each id is
isolated too — one order the broker will not answer about must not stop the
rest, which is the same per-item isolation discovery uses.
Returns the broker order ids settled, for the cycle report.
"""
order_ids = trade_repository.unsettled_broker_order_ids()
if not order_ids:
return []
try:
still_open = {order.order_id for order in broker.get_open_orders()}
except TraderError as exc:
_logger.error(
"could not read open orders: %s; not settling any terminal orders this cycle",
exc,
)
return []
candidates = [order_id for order_id in order_ids if order_id not in still_open]
if len(candidates) > _MAX_SETTLE_LOOKUPS:
_logger.warning(
"%s unsettled order(s) to check but only %s lookup(s) allowed per "
"cycle; checking the oldest %s, the rest next cycle",
len(candidates),
_MAX_SETTLE_LOOKUPS,
_MAX_SETTLE_LOOKUPS,
)
candidates = candidates[:_MAX_SETTLE_LOOKUPS]
settled: list[str] = []
for order_id in candidates:
try:
order = broker.get_order_by_id(order_id)
except TraderError as exc:
_logger.warning(
"could not read order %s: %s; leaving its trades row unsettled",
order_id,
exc,
)
continue
if order is None:
# The broker denies knowing this id. That is not evidence the order
# never filled — this app recorded it because the broker accepted it
# once — so it is reported and left alone rather than settled.
_logger.warning(
"the broker has no record of order %s, which this app did "
"place; leaving its trades row unsettled, because 'not found' "
"is not evidence that it never filled",
order_id,
)
continue
if order.status not in _TERMINAL_NON_FILLING_STATUSES:
continue
if order.filled_qty is None:
_logger.warning(
"%s: order %s is %s but the broker reported no filled "
"quantity; leaving its trades row unsettled, because unknown "
"is not zero and settling it would hide any shares it moved",
order.symbol,
order_id,
order.status,
)
continue
if order.filled_qty > 0:
# Real shares moved before this order ended. `reconcile_fills` still
# has to see it, and it skips anything already settled — so the
# window stays pinned by this row on purpose, and `mark_settled`
# stamps it once the round trip commits.
_logger.info(
"%s: order %s ended %s having filled %s share(s); leaving its "
"trades row unsettled so the fill is still reconciled",
order.symbol,
order_id,
order.status,
order.filled_qty,
)
continue
if trade_repository.mark_settled([order_id], now):
settled.append(order_id)
_logger.info(
"%s: order %s ended %s with no fill; settled its trades row so "
"it stops holding the fill-poll window open",
order.symbol,
order_id,
order.status,
)
return settled
[docs]
def reconcile_fills(
*,
broker,
outcome_repository,
trade_repository,
previous_positions: dict[str, Position],
now: datetime,
) -> list[str]:
"""Record fills, and open or close round trips as positions change.
Runs at the *start* of a cycle, so a fill that happened while nothing was
watching is on record before anything reasons about it. That in-between
state — an order placed but not yet filled — is precisely the one that no
Plan B fixture represented and that produced a live duplicate order.
Driven by the filled orders themselves, replayed oldest-first, rather than
by diffing position snapshots: a limit buy that fills and a trailing stop
that exits the same position before the *next* cycle even runs is a
symbol present in neither `previous_positions` nor `current`, so a
snapshot diff would never see the round trip at all — every fill would
still land on its `Trade` row, but no `RoundTrip` would ever exist.
Current positions are still read, and used for two things only: sourcing
a better price than an order without one carries, and a reconciliation
check afterwards that logs a mismatch rather than silently leaving a
round trip open (or absent) forever.
`get_filled_orders()` carries no dedupe of its own — the same closed order
reappears on every later cycle until it ages out of the broker's response
window — so every order is checked for prior settlement before it is
allowed to affect round-trip state at all. Without that, a replayed
buy/sell pair would open and close a second round trip for exactly the
same trade every idle cycle, silently inflating `total_realized_pl()`
without bound. The check is `trades.settled_at`, stamped on *every* fill
committed on a side; `OutcomeRepository.trade_settled` only knows the
first fill on each side, which left a replayed second buy free to open a
position-less ghost trip that then wedged the symbol forever.
A fill counts when shares actually moved (`_shares_moved`), not when the
terminal status string happens to be `"filled"`: a DAY entry that fills 20
of 29 shares and then expires put real shares in the account.
A side can also span more than one order (Alpaca can split a single exit
across several fills that arrive as separate orders): fills on the same
side are accumulated and committed as one `open_round_trip`/
`close_round_trip` call with a quantity-weighted average price, rather
than letting the first fill on a side decide the whole trip and silently
discarding the rest.
Never raises: a broker hiccup here must not cost the whole cycle.
"""
try:
filled = broker.get_filled_orders(since=_fill_poll_since(trade_repository, now))
current = {p.symbol: p for p in broker.get_positions()}
except TraderError as exc:
_logger.error("could not reconcile fills: %s", exc)
return []
ours: list[SubmittedOrder] = []
for order in filled:
if not _shares_moved(order):
continue
trade = trade_repository.get_by_order_id(order.order_id)
if trade is None:
# Not ours: hand-placed, or from before this app existed. Note
# that protective trailing stops *are* ours — `run_once` records
# every one it places — which is what lets a stop-driven exit,
# this app's dominant exit path, close its round trip at all.
continue
# Already reflected in round-trip state; replaying it would open or
# close a second trip for the same trade. `settled_at` is the
# exhaustive check (every fill committed on a side is stamped);
# `trade_settled` is kept for rows written before that column existed,
# where only the first fill on each side is discoverable.
if trade.settled_at is not None or outcome_repository.trade_settled(trade.id):
continue
ours.append(order)
# Oldest first: a round trip that opens and closes entirely inside one
# gap between cycles must be replayed buy-then-sell, and the broker's
# response order is not documented as chronological.
ours.sort(key=lambda order: (order.filled_at or order.submitted_at, order.order_id))
for order in ours:
outcome_repository.apply_fill(
order,
_fill_quantity(order),
_fill_price_for_ledger(order),
_fill_time(order, now),
)
by_symbol: dict[str, list[SubmittedOrder]] = {}
for order in ours:
by_symbol.setdefault(order.symbol, []).append(order)
changed: list[str] = []
for symbol in sorted(set(by_symbol) | set(previous_positions) | set(current)):
is_open = outcome_repository.open_round_trip_for(symbol) is not None
# A symbol whose round trip both opened and closed within this same
# call (the same-gap case) must appear once in `changed`, not twice.
symbol_changed = False
# Orders on the current, not-yet-committed side of the trip. Buys
# accumulate here while the trip isn't open yet; sells accumulate
# here once it is, until enough has sold to go flat.
pending_entry: list[SubmittedOrder] = []
pending_exit: list[SubmittedOrder] = []
# Whether any exit fill for this symbol landed in this same batch.
# Decided up front because `commit_entry` may run before the loop
# reaches the sell that reduced the position.
exits_this_call = any(
o.side is not OrderSide.BUY for o in by_symbol.get(symbol, [])
)
def commit_entry(*, closing: bool = False) -> None:
"""Open the round trip these accumulated buys represent.
`closing` marks the call made from the sell branch, where an exit
fill in this same batch explains a position that is already gone.
"""
nonlocal is_open, symbol_changed
if not pending_entry:
return
position = current.get(symbol)
if position is None and not closing:
# Buys with no position behind them and no exit in this batch
# to explain where the shares went. Opening a trip here
# creates a position-less ghost that can never close — and
# because an open trip makes every later buy on the symbol get
# discarded as pyramiding, that ghost permanently kills the
# symbol's outcome data. The realistic source is a replayed
# non-first buy of an already-closed split entry, which
# nothing else screens out. Left unsettled deliberately: if
# the position was merely not visible yet, the next cycle
# sees the fill again.
_logger.warning(
"%s: %s buy fill(s) (%s) have no position behind them and "
"no exit this cycle to explain one; not opening a round "
"trip, which would be a ghost that can never close",
symbol,
len(pending_entry),
", ".join(o.order_id for o in pending_entry),
)
pending_entry.clear()
return
first = pending_entry[0]
trade = trade_repository.get_by_order_id(first.order_id)
# The broker's own reported quantity and average entry price are
# more authoritative than what these orders alone add up to: they
# reflect every fill on the broker's side, including ones outside
# whatever window this call can see, and `avg_entry_price` is
# already the quantity-weighted figure the design calls for.
#
# But only while they still describe the *entry*. An exit fill in
# this same batch has already reduced the position, so preferring
# the broker's number there records the *remainder* as the entry:
# a 10-share buy followed by a 4-share exit in one gap recorded a
# 6-share entry and a realized P/L of 139.99 against a true
# 260.00. That case is ordinary now that trailing-stop fills are
# visible at all, so when an exit is present this sums the entry
# fills instead rather than compounding two estimates.
from_broker = position is not None and not exits_this_call
outcome_repository.open_round_trip(
symbol=symbol,
strategy_id=trade.strategy_id if trade else None,
entry_trade_id=trade.id if trade else None,
decision_id=None,
quantity=(
position.quantity
if from_broker
else sum((_fill_quantity(o) for o in pending_entry), Decimal(0))
),
entry_price=(
position.avg_entry_price
if from_broker
else _weighted_price(pending_entry, None)
),
opened_at=_fill_time(first, now),
)
is_open = True
symbol_changed = True
# Every contributing fill, not only the one `entry_trade_id` can
# name, so the broker replaying the rest cannot re-open anything.
trade_repository.mark_settled([o.order_id for o in pending_entry], now)
pending_entry.clear()
def commit_exit() -> None:
nonlocal is_open, symbol_changed
if not pending_exit:
return
first = pending_exit[0]
last = pending_exit[-1]
trade = trade_repository.get_by_order_id(first.order_id)
previous = previous_positions.get(symbol)
position_price = previous.current_price if previous is not None else None
trip = outcome_repository.open_round_trip_for(symbol)
# `trip` should always exist here (commit_exit is only reached
# once a trip is open), but a missing one is treated as "nothing
# to cap against" rather than crashing the cycle over it.
raw_sold = sum((_fill_quantity(o) for o in pending_exit), Decimal(0))
cap = trip.quantity if trip is not None else raw_sold
if raw_sold > cap:
# An entry recorded as smaller than what actually got sold —
# most likely a split entry whose second order arrived in a
# later cycle and was discarded as "already open, not
# pyramiding" (a documented, not fixed, limitation: see
# reconcile_fills's docstring). Capping keeps this exit's
# quantity and P/L honest against what this app actually
# believes it holds, rather than silently overstating both.
_logger.warning(
"%s: exit fills totaled %s shares against a %s-share "
"recorded entry; capping the recorded exit at %s rather "
"than overstating the position and its P/L",
symbol,
raw_sold,
cap,
cap,
)
quantity, exit_price = _capped_weighted_exit(
pending_exit, cap, position_price
)
closed = outcome_repository.close_round_trip(
symbol=symbol,
exit_trade_id=trade.id if trade else None,
quantity=quantity,
exit_price=exit_price,
closed_at=_fill_time(last, now),
)
is_open = False
if closed is not None:
symbol_changed = True
# Settled only now, at commit — including any fill dropped by the
# cap, which has been accounted for. A partial exit that never
# reaches this point stays unsettled on purpose, so the next cycle
# re-accumulates it from scratch.
trade_repository.mark_settled([o.order_id for o in pending_exit], now)
pending_exit.clear()
for order in by_symbol.get(symbol, []):
if order.side is OrderSide.BUY:
if is_open:
# This app does not pyramid: a buy fill while a trip is
# already open is not a second position. But it could
# also be the second half of a *split* entry whose first
# order already committed the trip in an earlier cycle —
# that case is a documented, not fixed, limitation (see
# reconcile_fills's docstring): this fill is discarded,
# so the trip's recorded quantity permanently understates
# what the broker actually holds. The exit-side cap this
# module also has exists specifically to keep that
# eventual understatement from corrupting P/L further.
_logger.warning(
"%s: buy fill %s arrived with a round trip already "
"open; ignoring rather than pyramiding. If this was "
"really the rest of a split entry, the recorded "
"quantity now understates the true position",
symbol,
order.order_id,
)
continue
pending_entry.append(order)
else:
if not is_open and not pending_entry:
# Not ours to close: either this app never opened it, or
# it was already closed by an earlier fill this cycle.
_logger.warning(
"%s: sell fill %s arrived with no round trip open; "
"ignoring rather than inventing one",
symbol,
order.order_id,
)
continue
# A sell means the entry phase (if a same-cycle entry was
# still accumulating) is over — commit it so there is a trip
# to sell against. `closing=True`: this exit is what accounts
# for the position those buys opened, so a symbol that is
# already flat is expected here rather than a ghost.
commit_entry(closing=True)
pending_exit.append(order)
trip = outcome_repository.open_round_trip_for(symbol)
sold_so_far = sum((_fill_quantity(o) for o in pending_exit), Decimal(0))
if trip is not None and sold_so_far >= trip.quantity:
commit_exit()
# Otherwise this is a partial exit that hasn't reached flat
# yet within this call; it stays pending (dropped at the end
# of this symbol's processing) and is recomputed from
# scratch next cycle, since none of its orders have been
# marked settled yet.
# Buys with no sell to trigger a commit this cycle still open the
# trip — the entry may finish in one order or several, but either
# way it should not wait for a sell that may never come.
commit_entry()
if pending_exit:
# A partial exit that hasn't reached the recorded entry quantity
# yet within this call. Correct as-is — nothing is corrupted, and
# none of these orders are marked settled, so the next cycle
# re-fetches and re-accumulates them from scratch — but silent
# otherwise, leaving an operator with no signal that a round trip
# is sitting part-way through an exit.
trip = outcome_repository.open_round_trip_for(symbol)
sold_so_far = sum((_fill_quantity(o) for o in pending_exit), Decimal(0))
_logger.warning(
"%s: %s of %s shares sold this cycle; the round trip remains "
"open and will re-accumulate the rest next cycle",
symbol,
sold_so_far,
trip.quantity if trip is not None else "an unknown number of",
)
if symbol_changed:
changed.append(symbol)
# Reconciliation: does the round-trip state this loop just settled on
# agree with what the broker reports right now? A mismatch means a
# fill this app could not see — hand-placed, from before this app
# existed, or outside whatever window `get_filled_orders` covers —
# changed the position. Logged rather than "fixed": fabricating a
# trip (or a close) from a price this app never observed would be
# worse than an honest, visible gap.
if is_open and symbol not in current:
_logger.warning(
"%s: a round trip is open on record but the broker reports "
"no position; it may have closed on a fill this app never saw",
symbol,
)
elif not is_open and symbol in current:
_logger.warning(
"%s: the broker reports an open position with no round trip "
"on record; it may have opened on a fill this app never saw",
symbol,
)
return changed