"""Repository for the shadow-portfolio simulator's five tables (issue #33).
One session per call, matching every other repository in this package — the
simulator is not performance-critical (it steps at most a handful of times
per real trading day, per strategy), so consistency with the rest of the
codebase's style matters more than batching writes into fewer transactions.
`simulation/step.py` reads a value (e.g. a position's current
`high_water_mark`) via one call, decides what to do with it, then writes the
result via a second call — deliberately, so the "test the stop against the
*prior* high-water mark" ordering the design spec requires falls out of the
caller's own control flow rather than needing an in-repository transaction.
"""
from datetime import datetime
from decimal import Decimal
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, sessionmaker
from trader.persistence.models import (
SimEquitySnapshot,
SimOrder,
SimPortfolio,
SimPosition,
SimRoundTrip,
)
__all__ = ["ABB_PORTFOLIO_SUFFIX", "SimulationRepository"]
#: `SimOrder.status` values.
ORDER_PENDING = "pending"
ORDER_FILLED = "filled"
ORDER_CANCELLED = "cancelled"
#: `SimRoundTrip.exit_reason` values.
EXIT_TRAILING_STOP = "trailing_stop"
EXIT_STRATEGY_SELL = "strategy_sell"
#: A position closed by the simulator's own eviction mirror (issue #83), not
#: by the trailing stop or a strategy's own sell signal. Kept distinct so
#: `trader sim-report`'s "return excluding stop exits" computation — anything
#: `!= EXIT_TRAILING_STOP` — continues to include it with no code change,
#: while a reader who cares can still tell an eviction from a genuine
#: strategy-driven exit.
EXIT_EVICTED = "eviction"
#: Suffix appended to a real `strategy_id` to build the Always-be-Buying
#: variant's portfolio identity (issue #42) — its own `SimPortfolio` row,
#: reading the *same* `strategy_id`'s recorded decisions but with sell
#: signals ignored, so only the trailing stop can ever close a position.
#: `sim_portfolios.strategy_id` has no format constraint, so this is a
#: convention, not a database rule: `"::"` cannot appear in a bare YAML
#: scalar id without quoting (`config/strategies.yaml`'s `id: str` has no
#: character restriction of its own), and no configured strategy uses it,
#: so in practice this suffix cannot collide with a real strategy id.
ABB_PORTFOLIO_SUFFIX = "::abb"
[docs]
class SimulationRepository:
"""Writes and reads the `sim_*` tables."""
def __init__(self, session_factory: sessionmaker[Session]) -> None:
self._session_factory = session_factory
# --- portfolios ----------------------------------------------------
[docs]
def get_or_create_portfolio(
self, strategy_id: str, *, starting_cash: Decimal
) -> SimPortfolio:
"""The portfolio for `strategy_id`, creating it (seeded at
`starting_cash`) on first use. `starting_cash` is ignored on every
call after the first — a config edit must not silently reset an
accumulating ledger."""
with self._session_factory() as session:
existing = session.scalars(
select(SimPortfolio).where(SimPortfolio.strategy_id == strategy_id)
).first()
if existing is not None:
return existing
portfolio = SimPortfolio(
strategy_id=strategy_id,
starting_cash=starting_cash,
cash=starting_cash,
last_bar_ts=None,
)
session.add(portfolio)
session.commit()
session.refresh(portfolio)
return portfolio
[docs]
def update_cash(self, portfolio_id: int, cash: Decimal) -> None:
with self._session_factory() as session:
portfolio = session.get(SimPortfolio, portfolio_id)
if portfolio is None:
return
portfolio.cash = cash
session.commit()
[docs]
def update_last_bar_ts(self, portfolio_id: int, bar_ts: datetime) -> None:
with self._session_factory() as session:
portfolio = session.get(SimPortfolio, portfolio_id)
if portfolio is None:
return
portfolio.last_bar_ts = bar_ts
session.commit()
[docs]
def all_portfolios(self) -> list[SimPortfolio]:
with self._session_factory() as session:
return list(session.scalars(select(SimPortfolio)))
# --- positions -------------------------------------------------------
[docs]
def get_position(self, portfolio_id: int, symbol: str) -> SimPosition | None:
with self._session_factory() as session:
return session.scalars(
select(SimPosition).where(
SimPosition.portfolio_id == portfolio_id,
SimPosition.symbol == symbol,
)
).first()
[docs]
def all_positions(self, portfolio_id: int) -> list[SimPosition]:
with self._session_factory() as session:
return list(
session.scalars(
select(SimPosition).where(SimPosition.portfolio_id == portfolio_id)
)
)
[docs]
def open_position(
self,
portfolio_id: int,
symbol: str,
*,
quantity: Decimal,
entry_price: Decimal,
high_water_mark: Decimal,
last_price: Decimal,
opened_at: datetime,
entry_confidence: float | None = None,
) -> SimPosition:
"""`entry_confidence` (issue #83) defaults to `None` for every caller
that predates it — an intentional "unknown", never a guessed value —
and is read once here, at fill time, from the decision that caused
this order (see `simulation/step.py`'s own lookup at the fill step)."""
with self._session_factory() as session:
position = SimPosition(
portfolio_id=portfolio_id,
symbol=symbol,
quantity=quantity,
entry_price=entry_price,
high_water_mark=high_water_mark,
last_price=last_price,
opened_at=opened_at,
entry_confidence=entry_confidence,
)
session.add(position)
session.commit()
session.refresh(position)
return position
[docs]
def mark_position(
self, position_id: int, *, high_water_mark: Decimal, last_price: Decimal
) -> None:
"""Write a new mark. The caller reads the *prior* value before
calling this — this method never returns the old one, on purpose,
so there is no temptation to test a breach against a value this
same call just overwrote."""
with self._session_factory() as session:
position = session.get(SimPosition, position_id)
if position is None:
return
position.high_water_mark = high_water_mark
position.last_price = last_price
session.commit()
[docs]
def close_position(self, position_id: int) -> None:
with self._session_factory() as session:
position = session.get(SimPosition, position_id)
if position is None:
return
session.delete(position)
session.commit()
# --- orders ------------------------------------------------------------
[docs]
def create_order(
self,
portfolio_id: int,
symbol: str,
*,
decision_id: int | None,
limit_price: Decimal,
quantity: Decimal,
created_bar_ts: datetime,
) -> SimOrder:
with self._session_factory() as session:
order = SimOrder(
portfolio_id=portfolio_id,
decision_id=decision_id,
symbol=symbol,
limit_price=limit_price,
quantity=quantity,
created_bar_ts=created_bar_ts,
status=ORDER_PENDING,
)
session.add(order)
session.commit()
session.refresh(order)
return order
[docs]
def pending_orders(self, portfolio_id: int) -> list[SimOrder]:
with self._session_factory() as session:
return list(
session.scalars(
select(SimOrder).where(
SimOrder.portfolio_id == portfolio_id,
SimOrder.status == ORDER_PENDING,
)
)
)
[docs]
def fill_order(self, order_id: int) -> None:
with self._session_factory() as session:
order = session.get(SimOrder, order_id)
if order is None:
return
order.status = ORDER_FILLED
session.commit()
[docs]
def cancel_order(self, order_id: int) -> None:
with self._session_factory() as session:
order = session.get(SimOrder, order_id)
if order is None:
return
order.status = ORDER_CANCELLED
session.commit()
# --- round trips ---------------------------------------------------
[docs]
def record_round_trip(
self,
portfolio_id: int,
symbol: str,
*,
decision_id: int | None,
quantity: Decimal,
entry_price: Decimal,
exit_price: Decimal,
realized_pl: Decimal,
opened_at: datetime,
closed_at: datetime,
exit_reason: str,
) -> SimRoundTrip:
with self._session_factory() as session:
trip = SimRoundTrip(
portfolio_id=portfolio_id,
decision_id=decision_id,
symbol=symbol,
quantity=quantity,
entry_price=entry_price,
exit_price=exit_price,
realized_pl=realized_pl,
opened_at=opened_at,
closed_at=closed_at,
exit_reason=exit_reason,
)
session.add(trip)
session.commit()
session.refresh(trip)
return trip
[docs]
def round_trips(self, portfolio_id: int) -> list[SimRoundTrip]:
with self._session_factory() as session:
return list(
session.scalars(
select(SimRoundTrip)
.where(SimRoundTrip.portfolio_id == portfolio_id)
.order_by(SimRoundTrip.closed_at.asc())
)
)
# --- equity snapshots ------------------------------------------------
[docs]
def snapshot_equity(
self, portfolio_id: int, bar_ts: datetime, equity: Decimal
) -> bool:
"""Insert one equity snapshot. Returns whether it was inserted.
`False` (not an exception) on the `(portfolio_id, bar_ts)` unique
constraint firing — the second of the step's two idempotency
mechanisms, alongside `SimPortfolio.last_bar_ts`: a step re-asked to
apply an already-applied bar must no-op, not double the curve.
"""
with self._session_factory() as session:
session.add(
SimEquitySnapshot(portfolio_id=portfolio_id, bar_ts=bar_ts, equity=equity)
)
try:
session.commit()
except IntegrityError:
session.rollback()
return False
return True
[docs]
def equity_curve(self, portfolio_id: int) -> list[SimEquitySnapshot]:
with self._session_factory() as session:
return list(
session.scalars(
select(SimEquitySnapshot)
.where(SimEquitySnapshot.portfolio_id == portfolio_id)
.order_by(SimEquitySnapshot.bar_ts.asc())
)
)