"""Repository for submitted orders (requirements §10).
A dry-run row is a real row: it records that the pipeline decided to act, and
what it would have sent. `alpaca_order_id` is NULL and `status` is `"dry_run"`,
so it can never be read as an order that reached a broker.
"""
from collections.abc import Sequence
from datetime import UTC, datetime
from decimal import Decimal
from sqlalchemy import func, select
from sqlalchemy.orm import Session, sessionmaker
from trader.domain import OrderSide, SubmittedOrder
from trader.persistence.models import Trade
__all__ = ["TradeRepository"]
[docs]
class TradeRepository:
"""Writes and reads the `trades` table."""
def __init__(self, session_factory: sessionmaker[Session]) -> None:
self._session_factory = session_factory
[docs]
def record_submission(
self, order: SubmittedOrder, strategy_id: str | None, is_paper: bool
) -> int:
"""Record an order the broker accepted. Returns the new row id.
`submitted_at` is the broker's timestamp, not ours: that is when the
order actually reached the market, whereas `now()` is when this process
got around to writing it down.
"""
trade = Trade(
ticker=order.symbol,
side=order.side.value,
quantity=order.quantity,
price=order.limit_price,
submitted_at=order.submitted_at,
strategy_id=strategy_id,
is_paper=is_paper,
alpaca_order_id=order.order_id,
status=order.status,
updated_at=datetime.now(UTC),
)
with self._session_factory() as session:
session.add(trade)
session.commit()
return trade.id
[docs]
def record_dry_run(
self,
symbol: str,
side: OrderSide,
quantity: int,
price: Decimal,
strategy_id: str | None,
is_paper: bool,
) -> int:
"""Record an order that was intended but deliberately not submitted."""
now = datetime.now(UTC)
trade = Trade(
ticker=symbol,
side=side.value,
quantity=Decimal(quantity),
price=price,
submitted_at=now,
strategy_id=strategy_id,
is_paper=is_paper,
alpaca_order_id=None,
status="dry_run",
updated_at=now,
)
with self._session_factory() as session:
session.add(trade)
session.commit()
return trade.id
[docs]
def mark_settled(self, order_ids: Sequence[str], settled_at: datetime) -> int:
"""Stamp `settled_at` on every trade with one of these broker order ids.
Returns how many rows were stamped. Called once a side of a round trip
is committed, so that *every* contributing fill — not only the first,
which is all `round_trips.entry_trade_id`/`exit_trade_id` can name —
is skipped when the broker replays it. Deliberately not called at fill
time: a partial exit that has not yet reached flat must stay unsettled
so the next cycle re-accumulates it.
Already-stamped rows keep their original timestamp: settlement happens
once, and re-stamping would make the column drift forward the same way
`Trade.filled_at` once did.
"""
if not order_ids:
return 0
statement = select(Trade).where(Trade.alpaca_order_id.in_(list(order_ids)))
with self._session_factory() as session:
stamped = 0
for trade in session.scalars(statement):
if trade.settled_at is not None:
continue
trade.settled_at = settled_at
trade.updated_at = datetime.now(UTC)
stamped += 1
session.commit()
return stamped
[docs]
def oldest_unsettled_submitted_at(self) -> datetime | None:
"""When the oldest still-outstanding order was submitted.
This is the correct lower bound for `get_filled_orders(since=...)`, and
the reasoning is worth keeping next to the query: Alpaca's
`GetOrdersRequest(after=...)` filters on **submission** time, not fill
time. Bounding on the newest recorded fill instead — the obvious
choice — would exclude an order submitted days ago that fills today,
and that is precisely what a GTC trailing stop is. Trailing stops are
this app's dominant exit path, so missing one means a round trip that
never closes.
Unsettled rather than unfilled: `settled_at` is stamped when a fill is
folded into round-trip state, so a partially-exited position stays in
the window until it reaches flat.
Dry-run rows are excluded. They have no `alpaca_order_id`, so no broker
order can ever match them and none will ever settle — including them
would pin the window open permanently after a single
`run-once --dry-run`.
Returns `None` when nothing is outstanding, which is the steady state
and the case the bound exists for: the caller then asks for a short
recent window instead of the broker's whole recent history.
"""
statement = select(func.min(Trade.submitted_at)).where(
Trade.settled_at.is_(None),
Trade.alpaca_order_id.is_not(None),
)
with self._session_factory() as session:
return session.scalar(statement)
[docs]
def unsettled_broker_order_ids(self) -> list[str]:
"""Broker order ids for every still-outstanding row, oldest first.
The same population `oldest_unsettled_submitted_at()` takes its minimum
over — deliberately, because these two must agree about what "still
outstanding" means or issue #1's sweep would chase rows that do not
actually hold the window open, and miss the one that does. Dry-run rows
are excluded here for the same reason as there: they carry no
`alpaca_order_id`, so no broker order can ever match them and asking
about them is meaningless.
Oldest first, because the row with the earliest `submitted_at` is
exactly the one pinning the fill-poll window; a caller that has to bound
how many broker lookups it makes in one cycle then spends them where
they narrow the window.
"""
statement = (
select(Trade.alpaca_order_id)
.where(
Trade.settled_at.is_(None),
Trade.alpaca_order_id.is_not(None),
)
.order_by(Trade.submitted_at, Trade.id)
)
with self._session_factory() as session:
# The `if order_id` is not redundant with the `is_not(None)` above so
# much as narrowing for the type checker: the column is
# `str | None`, and the SQL predicate is what the behaviour rests on.
# Both were mutated away together to confirm the tests catch it.
return [order_id for order_id in session.scalars(statement) if order_id]
[docs]
def get(self, trade_id: int) -> Trade | None:
"""One trade by id."""
with self._session_factory() as session:
return session.get(Trade, trade_id)
[docs]
def get_by_order_id(self, order_id: str) -> Trade | None:
"""The trade recorded for a broker order id, if this app placed it."""
statement = select(Trade).where(Trade.alpaca_order_id == order_id)
with self._session_factory() as session:
return session.scalars(statement).first()
[docs]
def exists_for_order(self, order_id: str) -> bool:
"""Whether this app placed the order with this broker id."""
return self.get_by_order_id(order_id) is not None
[docs]
def all(self) -> Sequence[Trade]:
"""Every trade, oldest first."""
statement = select(Trade).order_by(Trade.id)
with self._session_factory() as session:
return list(session.scalars(statement))
[docs]
def recent(self, limit: int = 20) -> list[Trade]:
"""The most recently submitted trades, newest first.
For the chat CLI (issue #13), which wants "what did you trade
recently" without pulling the whole table through `all()` and
reversing it client-side. Same ordering convention as
`DecisionRepository.recent` and `OutcomeRepository.recent`:
`submitted_at` descending with `id` descending as the tiebreak, since
two rows can share a timestamp.
"""
statement = (
select(Trade)
.order_by(Trade.submitted_at.desc(), Trade.id.desc())
.limit(limit)
)
with self._session_factory() as session:
return list(session.scalars(statement))
[docs]
def count(self) -> int:
"""How many trades have been recorded."""
with self._session_factory() as session:
return session.scalar(select(func.count()).select_from(Trade)) or 0