"""SQLAlchemy models for every table in requirements §10.
All tables are declared now so later slices add repositories without a
migration step. Slice 1 only wires the snapshot repository.
"""
from datetime import UTC, date, datetime
from decimal import Decimal
from sqlalchemy import (
Boolean,
Date,
DateTime,
Float,
ForeignKey,
Index,
Integer,
String,
Text,
TypeDecorator,
UniqueConstraint,
text,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
__all__ = [
"AccountSnapshot",
"AnalystConsensusHistory",
"ArchivedNews",
"BacktestRun",
"BacktestTrade",
"BarCoverage",
"Base",
"CachedBar",
"DecimalText",
"Decision",
"DeployedVersion",
"FocusList",
"IndexMembership",
"NewsArchiveCoverage",
"PositionSnapshot",
"RoundTrip",
"SimEquitySnapshot",
"SimOrder",
"SimPortfolio",
"SimPosition",
"SimRoundTrip",
"Trade",
"UtcDateTime",
"WatchlistEvent",
]
[docs]
class Base(DeclarativeBase):
"""Declarative base for all models."""
[docs]
class DecimalText(TypeDecorator):
"""Store `Decimal` as TEXT so SQLite cannot round-trip it through float.
SQLAlchemy's `Numeric` on SQLite goes via float and silently loses cents,
which is unacceptable in a trading ledger.
This column type is the single chokepoint enforcing "money is never a
float", so it *rejects* a float rather than laundering it: `Decimal(10.1)`
is `Decimal('10.0999999999999996447286321199499070644378662109375')`, which
would round-trip as a legitimate-looking value.
"""
impl = String
cache_ok = True
[docs]
def process_bind_param(self, value: Decimal | None, dialect: object) -> str | None:
"""Reject anything that is not a `Decimal` (or a plain `int`) before storing.
Raises:
TypeError: `value` is a `float` or any other non-`Decimal`,
non-`int` type — including `bool`, which Python treats as an
`int` subclass and which is excluded explicitly so a
stray boolean cannot silently pass as `0`/`1`.
"""
if value is None:
return None
# bool is an int subclass; exclude it explicitly.
if isinstance(value, bool) or not isinstance(value, (Decimal, int)):
raise TypeError(
f"money columns require Decimal, got {type(value).__name__}. "
"Storing a float here would silently lose precision."
)
return str(Decimal(value))
[docs]
def process_result_value(self, value: str | None, dialect: object) -> Decimal | None:
"""Rebuild the exact `Decimal` from the stored text, or `None`."""
return None if value is None else Decimal(value)
[docs]
class UtcDateTime(TypeDecorator):
"""Store tz-aware datetimes as naive UTC; return them tz-aware.
SQLite has no timezone support, so a plain `DateTime(timezone=True)`
column silently returns a naive value, and comparing that to an aware
`datetime` raises TypeError. Normalizing on the way in and re-attaching
UTC on the way out keeps every timestamp comparable.
Note the `astimezone` call: a non-UTC input must be *converted*, not merely
stripped, or its instant is silently corrupted.
"""
impl = DateTime
cache_ok = True
[docs]
def process_bind_param(
self, value: datetime | None, dialect: object
) -> datetime | None:
"""Convert (never merely strip) an aware datetime to naive UTC for storage.
Raises:
ValueError: `value` is naive (`tzinfo is None`) — guessing the
zone would silently corrupt the timeline, so a naive input is
refused rather than assumed to already be UTC.
"""
if value is None:
return None
if value.tzinfo is None:
raise ValueError(
"Refusing to store a naive datetime; pass a tz-aware UTC value. "
"Guessing the zone would silently corrupt the timeline."
)
return value.astimezone(UTC).replace(tzinfo=None)
[docs]
def process_result_value(
self, value: datetime | None, dialect: object
) -> datetime | None:
"""Re-attach UTC to the naive value read back from storage, or `None`."""
return None if value is None else value.replace(tzinfo=UTC)
[docs]
class Trade(Base):
"""An order the app submitted.
Every lifecycle column is declared here rather than added when first needed,
on purpose: `create_all` adds missing tables but never a missing column,
there is no Alembic, and this table goes from empty to populated in Slice 2
Plan B. A column not declared before the first order costs either Alembic or
the local history.
"""
__tablename__ = "trades"
__table_args__ = (
# The dedupe key for submissions. Without it a retried cycle could
# record the same broker order twice and double-count exposure. NULL is
# distinct from NULL in a SQLite unique index, so dry-run rows — which
# have no broker id — are unaffected.
UniqueConstraint("alpaca_order_id", name="uq_trades_alpaca_order_id"),
)
id: Mapped[int] = mapped_column(primary_key=True)
ticker: Mapped[str] = mapped_column(String(16), index=True)
side: Mapped[str] = mapped_column(String(8))
quantity: Mapped[Decimal] = mapped_column(DecimalText)
price: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
submitted_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
strategy_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
is_paper: Mapped[bool] = mapped_column(Boolean)
alpaca_order_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
status: Mapped[str | None] = mapped_column(String(32), nullable=True)
# Order lifecycle. Declared now, written later: `run-once` submits and does
# not poll for fills, so nothing in Slice 2 populates these. The daemon
# that does will not need a migration.
filled_qty: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
filled_avg_price: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
filled_at: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True)
updated_at: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True)
# When this fill was folded into round-trip state. `round_trips` can only
# name the *first* trade on each side (`entry_trade_id`/`exit_trade_id`),
# so a side built from several fills leaves the rest unreferenced — and
# `get_filled_orders()` keeps replaying them. A replayed unreferenced BUY
# used to open a brand-new, position-less "ghost" round trip that then
# wedged the symbol forever, because every genuine trip after it was
# discarded as pyramiding. Stamped at *commit* time, not at fill time, so
# a partial exit that has not yet reached flat stays unsettled and is
# deliberately re-accumulated next cycle.
settled_at: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True)
[docs]
class Decision(Base):
"""One strategy evaluation, whether or not it produced a trade."""
__tablename__ = "decisions"
id: Mapped[int] = mapped_column(primary_key=True)
decided_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
strategy_id: Mapped[str] = mapped_column(String(64), index=True)
ticker: Mapped[str] = mapped_column(String(16), index=True)
action: Mapped[str] = mapped_column(String(16))
# The decider's free-text rationale — WHY, always, never overwritten.
# Before issue #25, `run_once.py`'s buy path overwrote this with an
# execution summary at order time ("Bought 26 at limit 374.47..."),
# discarding the strategy's actual reason; the true rationale for those
# older rows survives only inside `inputs_json.raw_response.reason`
# (`ollama_news`) or is simply lost (a rule-based strategy has no
# `raw_response` to fall back to). Fixed going forward by `outcome_note`
# below taking over the execution-summary role.
reasoning: Mapped[str | None] = mapped_column(Text, nullable=True)
# What the decider *saw*, as against `reasoning`, which is what it *said*.
# Conflating the two makes it impossible to re-run a new prompt against the
# same input later and compare. Declared now and unused until the LLM
# slice, because this table stops being empty in Plan B and `create_all`
# cannot add a column afterwards.
inputs_json: Mapped[str | None] = mapped_column(Text, nullable=True)
trade_id: Mapped[int | None] = mapped_column(ForeignKey("trades.id"), nullable=True)
# Which gate produced this row's `action` when it diverged from the
# strategy's own signal (issue #21) — e.g. `price_floor`, `discovery_cap`,
# `total_exposure_cap`. Nullable and set only when relevant: most rows
# (a `buy`, a `sell`, an unvetoed `hold`) have nothing to name here.
# `run_once.py`'s `REASON_*`/`_REJECTION_*` constants and
# `discovery.filters`/`risk.guardrails`'s `REASON_*` constants are the
# closed set of values; never derived from `reasoning`'s free text, which
# is for a human and free to be reworded.
rejection_reason: Mapped[str | None] = mapped_column(String(32), nullable=True)
# WHAT happened — an execution outcome, distinct from WHY (`reasoning`
# above). Only set for a decision that actually executed an order ("Bought
# 26 at limit 374.47", "DRY RUN: would buy..."); `None` for every hold,
# rejection, or error, which have nothing to execute. Issue #25: these two
# facts used to share one column, and reading `reasoning` for a buy/sell
# answered "what happened", never "why", which is the opposite of what a
# later "why was this traded" review needs.
#
# Declared LAST, not next to `reasoning`, on purpose: SQLite's `ALTER
# TABLE ADD COLUMN` always appends, so `create_all` and the migration
# only render identical DDL (`test_migrations.py`) if the model's
# declaration order matches where the migration actually put it.
outcome_note: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
class WatchlistEvent(Base):
"""A ticker added to or removed from the watchlist, and by whom."""
__tablename__ = "watchlist_events"
id: Mapped[int] = mapped_column(primary_key=True)
occurred_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
ticker: Mapped[str] = mapped_column(String(16), index=True)
action: Mapped[str] = mapped_column(String(16))
# "user_fixed" or "claude_scan" — keeps the two lists distinguishable (§7).
source: Mapped[str] = mapped_column(String(32))
note: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
class AccountSnapshot(Base):
"""Periodic portfolio-level snapshot for performance tracking."""
__tablename__ = "account_snapshots"
id: Mapped[int] = mapped_column(primary_key=True)
captured_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
account_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
cash: Mapped[Decimal] = mapped_column(DecimalText)
equity: Mapped[Decimal] = mapped_column(DecimalText)
buying_power: Mapped[Decimal] = mapped_column(DecimalText)
portfolio_value: Mapped[Decimal] = mapped_column(DecimalText)
is_paper: Mapped[bool] = mapped_column(Boolean)
positions: Mapped[list["PositionSnapshot"]] = relationship(
back_populates="account_snapshot",
cascade="all, delete-orphan",
# Without this, iteration order is insertion order in practice but not
# SQL-guaranteed, so `positions[0]` is incidental rather than defined.
order_by="PositionSnapshot.id",
)
[docs]
class PositionSnapshot(Base):
"""One open position captured as part of an `AccountSnapshot`."""
__tablename__ = "position_snapshots"
id: Mapped[int] = mapped_column(primary_key=True)
account_snapshot_id: Mapped[int] = mapped_column(
ForeignKey("account_snapshots.id"), index=True
)
ticker: Mapped[str] = mapped_column(String(16), index=True)
quantity: Mapped[Decimal] = mapped_column(DecimalText)
avg_entry_price: Mapped[Decimal] = mapped_column(DecimalText)
current_price: Mapped[Decimal] = mapped_column(DecimalText)
market_value: Mapped[Decimal] = mapped_column(DecimalText)
unrealized_pl: Mapped[Decimal] = mapped_column(DecimalText)
account_snapshot: Mapped["AccountSnapshot"] = relationship(back_populates="positions")
[docs]
class BacktestRun(Base):
"""Parameters and results of a backtest (populated in Slice 4)."""
__tablename__ = "backtest_runs"
id: Mapped[int] = mapped_column(primary_key=True)
ran_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
strategy_id: Mapped[str] = mapped_column(String(64), index=True)
ticker: Mapped[str] = mapped_column(String(16))
start_date: Mapped[datetime] = mapped_column(UtcDateTime)
end_date: Mapped[datetime] = mapped_column(UtcDateTime)
parameters_json: Mapped[str | None] = mapped_column(Text, nullable=True)
starting_value: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
ending_value: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
total_return_pct: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
max_drawdown_pct: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
win_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
loss_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
# The do-nothing baseline for the same window, so a stored return stays
# interpretable years later without re-fetching bars and recomputing it by
# hand. Nullable because rows written before this column existed genuinely
# have no baseline — NULL says "never measured", which must stay
# distinguishable from a measured 0.00%.
benchmark_ending_value: Mapped[Decimal | None] = mapped_column(
DecimalText, nullable=True
)
benchmark_return_pct: Mapped[Decimal | None] = mapped_column(
DecimalText, nullable=True
)
[docs]
class CachedBar(Base):
"""One OHLCV bar, cached so backtests are fast, offline, and repeatable.
Yahoo revises history, so re-fetching would make a backtest's numbers drift
for reasons unrelated to the code. Caching pins the input.
`source` (issue #41) is `NULL` for every bar this app fetched itself
through `YFinanceProvider`/`BarCache` — the only live provider this app
has ever had, so an unlabelled row needs no label to be unambiguous.
A non-`NULL` value marks a row that arrived through `trader seed-bars`
instead: a named vendor tag (`seed_import.SOURCE_YFINANCE_SEED`,
`seed_import.SOURCE_IBKR_SEED`), never a boolean, because "seeded" alone
would hide that the two seed files come from two different vendors with
different adjustment/aggregation conventions and must stay distinguishable
from each other, not just from a live fetch. `BarRepository.delete_bars`
reads this column to refuse deleting a seeded row during a refresh — seed
data has no later corrected fetch coming the way a live tail-refresh
assumes, so a refresh silently replacing it with a differently-adjusted
live fetch would corrupt validated backtest input rather than correct it.
"""
__tablename__ = "bars"
__table_args__ = (
UniqueConstraint(
"symbol", "interval", "timestamp", name="uq_bars_symbol_interval_ts"
),
)
id: Mapped[int] = mapped_column(primary_key=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
interval: Mapped[str] = mapped_column(String(8))
timestamp: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
open: Mapped[Decimal] = mapped_column(DecimalText)
high: Mapped[Decimal] = mapped_column(DecimalText)
low: Mapped[Decimal] = mapped_column(DecimalText)
close: Mapped[Decimal] = mapped_column(DecimalText)
volume: Mapped[int] = mapped_column(Integer)
# Declared LAST, not next to `symbol`/`interval`, for the identical reason
# `Decision.outcome_note` is declared last: SQLite's `ALTER TABLE ADD
# COLUMN` always appends, so `create_all` and the migration only render
# identical DDL (`test_migrations.py`) if the model's declaration order
# matches where the migration actually put it.
source: Mapped[str | None] = mapped_column(String(32), nullable=True)
[docs]
class BarCoverage(Base):
"""Which date ranges have actually been fetched for a symbol.
Without this, a missing bar is ambiguous: the market may have been closed,
or the range may never have been downloaded. The second case would let a
backtest run across a hole and report confident numbers from partial data.
"""
__tablename__ = "bar_coverage"
id: Mapped[int] = mapped_column(primary_key=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
interval: Mapped[str] = mapped_column(String(8))
start_date: Mapped[datetime] = mapped_column(UtcDateTime)
end_date: Mapped[datetime] = mapped_column(UtcDateTime)
fetched_at: Mapped[datetime] = mapped_column(UtcDateTime)
[docs]
class ArchivedNews(Base):
"""One historical news article, kept so a decision can be replayed.
The live path deliberately does *not* read this table: it stores the news it
saw into `decisions.inputs_json`, because yfinance has no archive and the
snapshot is the only record that will ever exist. Alpaca (Benzinga) does have
history back to 2023, and that is what this table is for — accumulating
*backwards* so a prompt can be rebuilt as it would have looked on a past
date, through the same `build_messages` the live path uses.
Keyed `(symbol, published_at, article_id)`. All three are needed: Benzinga
publishes batches sharing a timestamp to the second, so the id is what keeps
the second story in a batch; and one article legitimately appears under
several symbols, filed once per symbol because that is how it was requested
and how the prompt attributes it.
Rows are immutable evidence of what was *said*, not of what was true. A
story corrected in place under the same id does not overwrite the row it
already has — the model's decision was made on the original wording, so
replacing it would change the question being replayed.
"""
__tablename__ = "news_archive"
__table_args__ = (
UniqueConstraint(
"symbol",
"published_at",
"article_id",
name="uq_news_archive_symbol_published_article",
),
# The one query the replay makes: everything for a symbol published
# before an instant. The unique constraint's index leads with the same
# two columns on SQLite, but naming this makes the read path's
# requirement explicit rather than a side effect of the key's order.
Index("ix_news_archive_symbol_published", "symbol", "published_at"),
)
id: Mapped[int] = mapped_column(primary_key=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
#: The provider's own id, as text. Alpaca's is an integer, but the archive is
#: keyed on this and `docs/ideas.md`'s fallback source for gaps
#: (web.archive.org) has no integer id to offer. Not nullable: SQLite treats
#: every NULL in a unique constraint as distinct, so a nullable id would
#: re-insert on every backfill instead of deduping.
article_id: Mapped[str] = mapped_column(String(64))
published_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
title: Mapped[str] = mapped_column(Text)
publisher: Mapped[str] = mapped_column(String(128))
summary: Mapped[str | None] = mapped_column(Text, nullable=True)
url: Mapped[str | None] = mapped_column(Text, nullable=True)
content_type: Mapped[str] = mapped_column(String(32))
#: When this row was archived. Not the same fact as `published_at`, and the
#: only way to tell a 2023 story fetched today from one fetched in 2023.
fetched_at: Mapped[datetime] = mapped_column(UtcDateTime)
[docs]
class NewsArchiveCoverage(Base):
"""Which spans of time have actually been fetched for a symbol.
Without this, "no rows" is two different claims: the symbol was quiet, or
nobody asked. Measured over 189 weeks, the thin names this app trades are
newsless in 41-73% of weeks (ITRG carried news in 51 weeks of 189), so a
backfill that failed halfway would read as a genuinely quiet stretch — and
the replay would then score the model on prompts missing news it would
really have had. This is `BarCoverage`'s reason, applied to news.
Spans are instants, deliberately unsnapped. `BarCoverage` snaps onto the bar
grid because a `1d` bar is midnight-dated; a news item has a real
publication time, and snapping would claim hours nobody fetched.
"""
__tablename__ = "news_archive_coverage"
id: Mapped[int] = mapped_column(primary_key=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
start_at: Mapped[datetime] = mapped_column(UtcDateTime)
end_at: Mapped[datetime] = mapped_column(UtcDateTime)
fetched_at: Mapped[datetime] = mapped_column(UtcDateTime)
[docs]
class BacktestTrade(Base):
"""One simulated fill.
Deliberately a separate table from `trades`: a simulated fill must never be
readable as an order that actually happened.
"""
__tablename__ = "backtest_trades"
id: Mapped[int] = mapped_column(primary_key=True)
backtest_run_id: Mapped[int] = mapped_column(
ForeignKey("backtest_runs.id"), index=True
)
ticker: Mapped[str] = mapped_column(String(16), index=True)
quantity: Mapped[Decimal] = mapped_column(DecimalText)
entry_at: Mapped[datetime] = mapped_column(UtcDateTime)
entry_price: Mapped[Decimal] = mapped_column(DecimalText)
exit_at: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True)
exit_price: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
pnl: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
[docs]
class RoundTrip(Base):
"""One completed position: opened, closed, and what it earned.
A round trip spans two trades, so realized P/L cannot live on either
`Trade` row without making the other misleading. Keeping it here makes
"what did this decision earn?" a lookup rather than a self-join.
Matched on the *position* going flat rather than on order pairs, because
Alpaca produces partial fills: a 29-share buy can fill 20 then 9, and an
order-pair rule mismatches every time that happens.
"""
__tablename__ = "round_trips"
__table_args__ = (
# Structural version of "one open trip per symbol": a convention is
# not enough, per the same reasoning as `trades.alpaca_order_id`'s
# unique constraint. Partial on BOTH dialects, because the
# constraint is only about *open* trips — many closed trips per
# symbol are normal and expected. `sqlite_where` alone (as this was
# originally written) is SQLite-dialect-specific: on Postgres,
# `create_all` silently produced an UNCONDITIONAL unique index on
# `symbol` with no WHERE clause at all, making every symbol
# representable at most ONCE in `round_trips`, ever — found via a
# migrated-database ETL's row-count mismatch (issue #26): 10 of 27
# real round trips were silently dropped by that unconditional
# index. `postgresql_where` is what closes the gap; see the
# corrective migration this shipped alongside.
Index(
"uq_round_trips_open_symbol",
"symbol",
unique=True,
sqlite_where=text("is_open"),
postgresql_where=text("is_open"),
),
)
id: Mapped[int] = mapped_column(primary_key=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
strategy_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
# The trade and decision that *opened* the position. Where several fills
# contributed, these reference the first. Valid because this app holds one
# position per symbol and never pyramids.
entry_trade_id: Mapped[int | None] = mapped_column(
ForeignKey("trades.id"), nullable=True
)
exit_trade_id: Mapped[int | None] = mapped_column(
ForeignKey("trades.id"), nullable=True
)
decision_id: Mapped[int | None] = mapped_column(
ForeignKey("decisions.id"), nullable=True
)
quantity: Mapped[Decimal] = mapped_column(DecimalText)
entry_price: Mapped[Decimal] = mapped_column(DecimalText)
exit_price: Mapped[Decimal] = mapped_column(DecimalText)
realized_pl: Mapped[Decimal] = mapped_column(DecimalText)
opened_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
closed_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
# Whether the position is still open. An explicit flag rather than
# inferring from a null exit price, because zero is a legitimate exit
# price and "unset" and "zero" must stay distinguishable.
is_open: Mapped[bool] = mapped_column(Boolean, default=True, index=True)
[docs]
class AnalystConsensusHistory(Base):
"""One `trader scan-universe` fetch of a symbol's analyst consensus.
Append-only, like `ArchivedNews` — no unique constraint, because a later
row is not a replacement, it is evidence the consensus moved. Freshness is
answered by querying the newest row per symbol
(`AnalystConsensusRepository.latest_for_symbols`), not by a separate
coverage table: `docs/ideas.md`'s "age-based freshness policy" model,
confirmed the right shape for non-bar data because `BarCoverage`/
`NewsArchiveCoverage` answer "has this span been fetched", and the
question here is "how old is the newest row", which has no span at all.
"""
__tablename__ = "analyst_consensus_history"
__table_args__ = (
Index("ix_analyst_consensus_history_symbol_fetched", "symbol", "fetched_at"),
)
id: Mapped[int] = mapped_column(primary_key=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
fetched_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
strong_buy: Mapped[int] = mapped_column(Integer)
buy: Mapped[int] = mapped_column(Integer)
hold: Mapped[int] = mapped_column(Integer)
sell: Mapped[int] = mapped_column(Integer)
strong_sell: Mapped[int] = mapped_column(Integer)
#: A rating (1..5), not money — `Float`, not `DecimalText`.
recommendation_mean: Mapped[float | None] = mapped_column(Float, nullable=True)
#: `True` when yfinance omitted `recommendationMean` and this app derived
#: it from the five counts instead. See `AnalystOpinion` for the same flag.
recommendation_mean_is_derived: Mapped[bool] = mapped_column(Boolean)
target_mean_price: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
target_high_price: Mapped[Decimal | None] = mapped_column(DecimalText, nullable=True)
#: e.g. `"yfinance"` — the only source today, named anyway so a second
#: source later does not need a migration to become distinguishable.
source: Mapped[str] = mapped_column(String(32))
[docs]
class IndexMembership(Base):
"""Which symbols `trader scan-universe` found in a named index.
A set, not a history — unlike `AnalystConsensusHistory`, membership on a
given day is either true or it is not, so re-running the scrape upserts
rather than appending. `UniqueConstraint` is what makes that upsert
meaningful rather than merely conventional.
"""
__tablename__ = "index_membership"
__table_args__ = (
UniqueConstraint("index_name", "symbol", name="uq_index_membership_index_symbol"),
)
id: Mapped[int] = mapped_column(primary_key=True)
#: e.g. `"sp500"`. `trader scan-universe` supports only that value today.
index_name: Mapped[str] = mapped_column(String(32), index=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
discovered_at: Mapped[datetime] = mapped_column(UtcDateTime)
[docs]
class FocusList(Base):
"""A symbol `trader build-focus-list` computed overnight for a session
(issue #44).
`trading_day` is a `Date`, not a `UtcDateTime` instant — the same
"a claim about the session grid, not about instants" lesson `BarCoverage`
already paid for. It is the session this row is *for*, stamped from the
broker clock's `next_open.date()` at the moment the batch runs (Decision
7), never wall-clock UTC midnight when the batch happened to execute.
Intraday, a cycle reads only today's `trading_day`; a batch that never
ran, or stamped the wrong day, degrades to zero rows for today — the same
shape `themes: []` and a stale `analyst_consensus_history` already
degrade to.
A history, not a pure set (unlike `IndexMembership`): yesterday's rows are
never deleted, only superseded by a different `trading_day`, so "what was
the app considering watching on a given date" stays answerable the same
way `watchlist_events` answers "why did it buy UUUU". `UNIQUE(trading_day,
symbol)` is what makes a same-day rerun (a manual retry after a partial
`build-focus-list` failure) an upsert rather than a duplicate.
`theme`/`origin` mirror `Candidate`'s own fields (today always
`"analyst_scan"`) rather than being hard-coded, so a future batch-only
candidate source needs no column change. `headline`/`url` are schema
symmetry with `Candidate` and are always `None` for an analyst_scan-origin
row — there is no headline behind an analyst-consensus signal.
`recommendation_mean` is a snapshot of the signal that earned the slot,
the same "why is this here" motivation `watchlist_events` already serves
for theme-search discoveries.
"""
__tablename__ = "focus_list"
__table_args__ = (
UniqueConstraint(
"trading_day", "symbol", name="uq_focus_list_trading_day_symbol"
),
)
id: Mapped[int] = mapped_column(primary_key=True)
trading_day: Mapped[date] = mapped_column(Date, index=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
theme: Mapped[str] = mapped_column(String(64))
origin: Mapped[str] = mapped_column(String(32))
headline: Mapped[str | None] = mapped_column(Text, nullable=True)
url: Mapped[str | None] = mapped_column(Text, nullable=True)
#: A rating (1..5), not money — `Float`, not `DecimalText`, same as
#: `AnalystConsensusHistory.recommendation_mean`.
recommendation_mean: Mapped[float | None] = mapped_column(Float, nullable=True)
generated_at: Mapped[datetime] = mapped_column(UtcDateTime)
[docs]
class DeployedVersion(Base):
"""A code version a running process actually observed itself on.
Sparse, not per-cycle: `DeployedVersionRepository.record_if_changed`
inserts only when the detected `git_sha` differs from the most recent
row, so a `trader run` that runs for days on unchanged code adds nothing
here — this is an event log of deploys actually observed, not a
per-invocation timestamp.
`trades`/`decisions`/`round_trips` are not joined to this table at write
time and get no new column: a row's version is found at query time as
the `DeployedVersion` with the greatest `detected_at` at or before that
row's own timestamp. A row that predates this table entirely has no
match and must read as unknown version — never as the earliest recorded
one, the same "unknown is not zero" discipline as the missing-analyst-
coverage check.
"""
__tablename__ = "deployed_versions"
id: Mapped[int] = mapped_column(primary_key=True)
#: Full SHA, never abbreviated — `versioning.UNKNOWN_GIT_SHA` when the
#: process could not determine it (not a git checkout, no `git` on
#: `PATH`), recorded explicitly rather than guessed.
git_sha: Mapped[str] = mapped_column(String(40))
#: `None` when no branch name resolves (detached `HEAD`, or the SHA
#: itself is unknown) — never a guess.
git_ref: Mapped[str | None] = mapped_column(String(255), nullable=True)
detected_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
[docs]
class SimPortfolio(Base):
"""One simulated portfolio per configured strategy (issue #33).
Forward-only and structurally isolated from real money: nothing in
`simulation/` imports `brokers/` or `execution/`, and no flag on `trades`
or `round_trips` marks a row as simulated — the isolation is that
simulated state has nowhere else to live but these five tables.
`last_bar_ts` is the idempotency guard: a cycle runs roughly every 15
minutes against a daily bar, so the step is *asked* to run far more often
than there is new data, and must apply each bar exactly once. `None`
means no bar has ever been applied yet, not "applied at time zero".
"""
__tablename__ = "sim_portfolios"
__table_args__ = (
UniqueConstraint("strategy_id", name="uq_sim_portfolios_strategy_id"),
)
id: Mapped[int] = mapped_column(primary_key=True)
strategy_id: Mapped[str] = mapped_column(String(64), index=True)
starting_cash: Mapped[Decimal] = mapped_column(DecimalText)
cash: Mapped[Decimal] = mapped_column(DecimalText)
last_bar_ts: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True)
[docs]
class SimOrder(Base):
"""A pending simulated limit buy (issue #33).
Mirrors the real pipeline's own unfilled-limit-buy state, which is
deliberately modelled rather than assumed away — filling instantly at
the current price would both overstate results (you always get in) and
erase the exact in-between state that let the real pipeline double-order
a symbol against the live account.
`decision_id` is nullable so a fill's provenance survives (which real
decision proposed this simulated buy), without forcing every order to
trace to one — a defensive `None` is cheaper than failing the whole step
over a lookup that should always succeed but is not load-bearing if it
doesn't.
"""
__tablename__ = "sim_orders"
id: Mapped[int] = mapped_column(primary_key=True)
portfolio_id: Mapped[int] = mapped_column(ForeignKey("sim_portfolios.id"), index=True)
decision_id: Mapped[int | None] = mapped_column(
ForeignKey("decisions.id"), nullable=True
)
symbol: Mapped[str] = mapped_column(String(16), index=True)
limit_price: Mapped[Decimal] = mapped_column(DecimalText)
quantity: Mapped[Decimal] = mapped_column(DecimalText)
created_bar_ts: Mapped[datetime] = mapped_column(UtcDateTime)
#: `"pending"` | `"filled"` | `"cancelled"`. A plain string, not an enum
#: column, matching `Trade.status`'s own convention elsewhere in this file.
status: Mapped[str] = mapped_column(String(16))
[docs]
class SimPosition(Base):
"""One open simulated position (issue #33).
`high_water_mark` is the trail's own ratcheted peak — tested against on
the *prior* value each step, per the design doc's pessimistic-ordering
rule, never the value just written this same step. `last_price` is
tracked separately from `high_water_mark` (issue #33's own addition
beyond the design doc's bare field list): an equity snapshot needs a
valuation for every open position on every stepped bar, including a bar
where this particular symbol had no trade of its own (a holiday, a thin
listing) and so was never marked or tested this step.
`entry_confidence` (issue #83) is the `signal_confidence` off the
`Decision` whose BUY opened this position — captured once, at fill time,
rather than re-derived later, because the position can outlive the bar
range a later query might search. `Float`, not `DecimalText`: it mirrors
`decisions.inputs_json["signal_confidence"]`/`Candidate.confidence`,
both plain `float` throughout this codebase (`llm/prompt.py` clamps every
model confidence into 0..1), never money. `None` when the opening
decision recorded no confidence (a rule strategy) or could not be found
— `simulation/step.py`'s own eviction mirror excludes such a holder from
consideration entirely, the same "unknown is not zero" rule
`pipeline/run_once.py`'s `_DiscoveredHolder` uses on the real path.
"""
__tablename__ = "sim_positions"
__table_args__ = (
UniqueConstraint(
"portfolio_id", "symbol", name="uq_sim_positions_portfolio_symbol"
),
)
id: Mapped[int] = mapped_column(primary_key=True)
portfolio_id: Mapped[int] = mapped_column(ForeignKey("sim_portfolios.id"), index=True)
symbol: Mapped[str] = mapped_column(String(16), index=True)
quantity: Mapped[Decimal] = mapped_column(DecimalText)
entry_price: Mapped[Decimal] = mapped_column(DecimalText)
high_water_mark: Mapped[Decimal] = mapped_column(DecimalText)
last_price: Mapped[Decimal] = mapped_column(DecimalText)
opened_at: Mapped[datetime] = mapped_column(UtcDateTime)
entry_confidence: Mapped[float | None] = mapped_column(Float, nullable=True)
[docs]
class SimRoundTrip(Base):
"""One closed simulated position: opened, closed, and what it earned
(issue #33).
`exit_reason` (`"trailing_stop"` | `"strategy_sell"`) is the whole point
of this table's existence separate from a bare P&L number: if a
strategy's returns come entirely from the trail firing, its entry logic
is worthless while its total P&L looks respectable, and nothing else
this app records would show that. `trader sim-report` reports both the
full ledger and a stop-excluded view from this one column.
"""
__tablename__ = "sim_round_trips"
id: Mapped[int] = mapped_column(primary_key=True)
portfolio_id: Mapped[int] = mapped_column(ForeignKey("sim_portfolios.id"), index=True)
decision_id: Mapped[int | None] = mapped_column(
ForeignKey("decisions.id"), nullable=True
)
symbol: Mapped[str] = mapped_column(String(16), index=True)
quantity: Mapped[Decimal] = mapped_column(DecimalText)
entry_price: Mapped[Decimal] = mapped_column(DecimalText)
exit_price: Mapped[Decimal] = mapped_column(DecimalText)
realized_pl: Mapped[Decimal] = mapped_column(DecimalText)
opened_at: Mapped[datetime] = mapped_column(UtcDateTime)
closed_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
exit_reason: Mapped[str] = mapped_column(String(16))
[docs]
class SimEquitySnapshot(Base):
"""The simulated equity curve, one row per stepped bar (issue #33).
What yields max drawdown and total return in `trader sim-report`. The
unique constraint on `(portfolio_id, bar_ts)` is the second of two
idempotency mechanisms (`SimPortfolio.last_bar_ts` is the first) — a step
asked to reapply an already-applied bar hits this constraint and no-ops
rather than doubling the curve.
"""
__tablename__ = "sim_equity_snapshots"
__table_args__ = (
UniqueConstraint(
"portfolio_id", "bar_ts", name="uq_sim_equity_snapshots_portfolio_bar"
),
)
id: Mapped[int] = mapped_column(primary_key=True)
portfolio_id: Mapped[int] = mapped_column(ForeignKey("sim_portfolios.id"), index=True)
bar_ts: Mapped[datetime] = mapped_column(UtcDateTime, index=True)
equity: Mapped[Decimal] = mapped_column(DecimalText)