87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
from dataclasses import dataclass
|
|
|
|
SIDES = ("BUY", "SELL")
|
|
|
|
|
|
@dataclass
|
|
class Trade:
|
|
trade_id: str
|
|
symbol: str
|
|
side: str
|
|
qty: int
|
|
price: float
|
|
|
|
def signed_qty(self):
|
|
if self.side == "SELL" and self.qty > 0:
|
|
self.qty = -self.qty
|
|
return self.qty
|
|
|
|
|
|
@dataclass
|
|
class Position:
|
|
symbol: str
|
|
qty: int = 0
|
|
avg_price: float = 0.0
|
|
realized: float = 0.0
|
|
|
|
def update(self, qty, price):
|
|
new_qty = self.qty + qty
|
|
if self.qty == 0 or (self.qty > 0) == (qty > 0):
|
|
self.avg_price = (self.avg_price * self.qty + price * qty) / new_qty
|
|
else:
|
|
closed = min(abs(qty), abs(self.qty))
|
|
direction = 1 if self.qty > 0 else -1
|
|
self.realized += closed * (price - self.avg_price) * direction
|
|
if new_qty == 0:
|
|
self.avg_price = 0.0
|
|
self.qty = new_qty
|
|
|
|
|
|
class Ledger:
|
|
def __init__(self, logger):
|
|
self.log = logger
|
|
self.positions = {}
|
|
self.history = []
|
|
|
|
def _validate(self, trade):
|
|
if trade.side not in SIDES:
|
|
raise ValueError(f"bad side {trade.side!r}")
|
|
if trade.qty <= 0:
|
|
raise ValueError(f"bad qty {trade.qty!r}")
|
|
if trade.price <= 0:
|
|
raise ValueError(f"bad price {trade.price!r}")
|
|
|
|
def apply(self, trade):
|
|
self.log.push(trade_id=trade.trade_id, symbol=trade.symbol)
|
|
try:
|
|
self._validate(trade)
|
|
except ValueError as exc:
|
|
self.log.error(f"rejected: {exc}")
|
|
raise
|
|
pos = self.positions.setdefault(trade.symbol, Position(trade.symbol))
|
|
pos.update(trade.signed_qty(), trade.price)
|
|
self.history.append(trade)
|
|
self.log.info(f"applied {trade.side} {abs(trade.qty)}@{trade.price:.2f} -> {pos.qty}")
|
|
return pos
|
|
|
|
def position(self, symbol):
|
|
return self.positions.get(symbol, Position(symbol))
|
|
|
|
@classmethod
|
|
def rebuild(cls, trades, logger):
|
|
ledger = cls(logger)
|
|
for trade in trades:
|
|
try:
|
|
ledger.apply(trade)
|
|
except ValueError:
|
|
continue
|
|
return ledger
|
|
|
|
def reconcile(self, other):
|
|
breaks = []
|
|
for symbol in sorted(set(self.positions) | set(other.positions)):
|
|
mine, theirs = self.position(symbol), other.position(symbol)
|
|
if mine.qty != theirs.qty or abs(mine.avg_price - theirs.avg_price) > 1e-9:
|
|
breaks.append((symbol, mine.qty, theirs.qty))
|
|
return breaks
|