from __future__ import annotations import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any from investly.domain import Action, Recommendation, RuleCitation, SimAction def _number(value: object, field_name: str) -> float: if not isinstance(value, (int, float)) or isinstance(value, bool): raise ValueError(f"invalid numeric portfolio field: {field_name}") return float(value) def _optional_number(value: object, field_name: str) -> float | None: if value is None: return None return _number(value, field_name) def _integer(value: object, field_name: str) -> int: if not isinstance(value, int) or isinstance(value, bool): raise ValueError(f"invalid integer portfolio field: {field_name}") return value @dataclass class Position: symbol: str quantity: int average_price: float active_recommendation_id: str | None = None target: float | None = None stop: float | None = None @dataclass class Portfolio: portfolio_id: str initial_cash: float cash: float positions: dict[str, Position] = field(default_factory=dict) max_position_pct: float = 0.20 high_confidence_position_pct: float = 1.00 concentration_confidence_threshold: float = 0.90 risk_per_trade_pct: float = 0.01 @classmethod def create(cls, portfolio_id: str, budget: float) -> "Portfolio": return cls(portfolio_id=portfolio_id, initial_cash=budget, cash=budget) def to_dict(self) -> dict[str, Any]: return { "portfolio_id": self.portfolio_id, "initial_cash": self.initial_cash, "cash": self.cash, "max_position_pct": self.max_position_pct, "high_confidence_position_pct": self.high_confidence_position_pct, "concentration_confidence_threshold": self.concentration_confidence_threshold, "risk_per_trade_pct": self.risk_per_trade_pct, "positions": { symbol: { "symbol": position.symbol, "quantity": position.quantity, "average_price": position.average_price, "active_recommendation_id": position.active_recommendation_id, "target": position.target, "stop": position.stop, } for symbol, position in self.positions.items() }, } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "Portfolio": raw_positions = payload.get("positions") positions: dict[str, Position] = {} if not isinstance(raw_positions, dict): raise ValueError("invalid portfolio positions payload") for symbol, raw in raw_positions.items(): if not isinstance(symbol, str) or not isinstance(raw, dict): raise ValueError("invalid portfolio position entry") active_id = raw.get("active_recommendation_id") if active_id is not None and not isinstance(active_id, str): raise ValueError(f"invalid position recommendation id: {symbol}") positions[symbol] = Position( symbol=symbol, quantity=_integer(raw.get("quantity"), f"positions.{symbol}.quantity"), average_price=_number(raw.get("average_price"), f"positions.{symbol}.average_price"), active_recommendation_id=active_id, target=_optional_number(raw.get("target"), f"positions.{symbol}.target"), stop=_optional_number(raw.get("stop"), f"positions.{symbol}.stop"), ) portfolio_id = payload.get("portfolio_id") if not isinstance(portfolio_id, str) or not portfolio_id: raise ValueError("invalid portfolio_id") return cls( portfolio_id=portfolio_id, initial_cash=_number(payload.get("initial_cash"), "initial_cash"), cash=_number(payload.get("cash"), "cash"), positions=positions, max_position_pct=_number(payload.get("max_position_pct", 0.20), "max_position_pct"), high_confidence_position_pct=_number( payload.get("high_confidence_position_pct", 1.00), "high_confidence_position_pct" ), concentration_confidence_threshold=_number( payload.get("concentration_confidence_threshold", 0.90), "concentration_confidence_threshold", ), risk_per_trade_pct=_number(payload.get("risk_per_trade_pct", 0.01), "risk_per_trade_pct"), ) def _action( self, recommendation_id: str, symbol: str, action: str, quantity: int, execution_price: float, cash_before: float, reason: str, rules: list[RuleCitation], before: dict[str, Any], ) -> SimAction: return SimAction( str(uuid.uuid4()), recommendation_id, self.portfolio_id, symbol, action, quantity, execution_price, cash_before, self.cash, reason, rules, datetime.now(timezone.utc), before, self.to_dict(), ) def _sell_all( self, *, recommendation_id: str, symbol: str, execution_price: float, action: str, reason: str, rules: list[RuleCitation], ) -> SimAction: before = self.to_dict() cash_before = self.cash position = self.positions.get(symbol) if position is None: return self._action( recommendation_id, symbol, "DO_NOTHING", 0, execution_price, cash_before, "exit instruction ignored because no position is held", rules, before, ) quantity = position.quantity self.cash += quantity * execution_price del self.positions[symbol] return self._action( recommendation_id, symbol, action, quantity, execution_price, cash_before, reason, rules, before, ) def apply(self, rec: Recommendation, execution_price: float | None = None) -> SimAction: before = self.to_dict() cash_before = self.cash fill_price = rec.reference_price if execution_price is None else execution_price if fill_price <= 0: raise ValueError("execution price must be positive") existing = self.positions.get(rec.symbol) if rec.action == Action.SELL: return self._sell_all( recommendation_id=rec.recommendation_id, symbol=rec.symbol, execution_price=fill_price, action="SELL", reason="engine SELL recommendation closed the full simulated position", rules=[*rec.rules, RuleCitation( "SIM.FOLLOW_RECOMMENDATION.001", 1, "simulation exits only from engine recommendation or active target/stop instruction", )], ) if rec.action == Action.HOLD: return self._action( rec.recommendation_id, rec.symbol, "HOLD" if existing is not None else "DO_NOTHING", 0, fill_price, cash_before, "engine HOLD recommendation preserved the existing position" if existing is not None else "HOLD ignored because no position is held", [*rec.rules, RuleCitation( "SIM.FOLLOW_RECOMMENDATION.001", 1, "simulation does not invent portfolio activity", )], before, ) if rec.action != Action.BUY or not rec.stop or fill_price <= rec.stop: return self._action( rec.recommendation_id, rec.symbol, "DO_NOTHING", 0, fill_price, cash_before, "recommendation not eligible for new buy at verified execution price", [RuleCitation( "SIM.ACTIONABILITY.001", 2, "only BUY recommendations with fill above invalidation can open or add positions", )], before, ) unit_risk = fill_price - rec.stop by_risk = int((self.initial_cash * self.risk_per_trade_pct) / unit_risk) if unit_risk > 0 else 0 existing_value = existing.quantity * fill_price if existing else 0.0 position_pct = ( self.high_confidence_position_pct if rec.confidence >= self.concentration_confidence_threshold else self.max_position_pct ) remaining_cap = max(0.0, self.initial_cash * position_pct - existing_value) by_cap = int(remaining_cap / fill_price) by_cash = int(self.cash / fill_price) qty = max(0, min(by_risk, by_cap, by_cash)) if qty < 1: return self._action( rec.recommendation_id, rec.symbol, "DO_NOTHING", 0, fill_price, cash_before, "risk/cash/concentration limits permit no additional whole-share position", [RuleCitation( "RISK.POSITION_SIZE.001", 2, "risk, confidence and concentration sizing", )], before, ) cost = qty * fill_price self.cash -= cost if existing is None: self.positions[rec.symbol] = Position( rec.symbol, qty, fill_price, rec.recommendation_id, rec.target, rec.stop, ) else: total_qty = existing.quantity + qty weighted_cost = ( existing.quantity * existing.average_price + qty * fill_price ) / total_qty self.positions[rec.symbol] = Position( rec.symbol, total_qty, weighted_cost, rec.recommendation_id, rec.target, rec.stop, ) concentration_reason = ( "confidence >=90% permits concentrated allocation; risk sizing still limits actual capital" if rec.confidence >= self.concentration_confidence_threshold else "normal diversified position ceiling applies below 90% confidence" ) return self._action( rec.recommendation_id, rec.symbol, "BUY", qty, fill_price, cash_before, "BUY executed at verified session open and sized by risk/capital limits", [ RuleCitation("RISK.POSITION_SIZE.001", 2, concentration_reason), RuleCitation("SIM.FILL.PRICE.001", 1, "paper fills use verified session open, never pre-open reference price"), RuleCitation("SIM.FOLLOW_RECOMMENDATION.001", 1, "simulation opens or adds only from an engine BUY recommendation"), ], before, ) def apply_intraday_exit( self, symbol: str, *, session_open: float, high: float, low: float, ) -> SimAction | None: position = self.positions.get(symbol) if position is None or position.active_recommendation_id is None: return None target_hit = position.target is not None and high >= position.target stop_hit = position.stop is not None and low <= position.stop if not target_hit and not stop_hit: return None if stop_hit: assert position.stop is not None price = session_open if session_open <= position.stop else position.stop return self._sell_all( recommendation_id=position.active_recommendation_id, symbol=symbol, execution_price=price, action="STOP_EXIT", reason=( "active recommendation stop/invalidation executed; stop wins when daily OHLC " "cannot resolve same-session target/stop ordering" ), rules=[RuleCitation( "SIM.EXIT.STOP.001", 1, "execute the active recommendation invalidation without hindsight", )], ) assert position.target is not None price = session_open if session_open >= position.target else position.target return self._sell_all( recommendation_id=position.active_recommendation_id, symbol=symbol, execution_price=price, action="TARGET_EXIT", reason="active recommendation target reached and simulated position closed", rules=[RuleCitation( "SIM.EXIT.TARGET.001", 1, "execute target from the active recommendation", )], ) def apply_session( self, rec: Recommendation, *, session_open: float, high: float, low: float, ) -> SimAction: """Apply the morning instruction and any same-session target/stop as one idempotent record.""" before = self.to_dict() cash_before = self.cash morning = self.apply(rec, execution_price=session_open) if morning.action == "SELL": morning.legs = [{"action": "SELL", "quantity": morning.quantity, "price": morning.price}] return morning exit_action = self.apply_intraday_exit( rec.symbol, session_open=session_open, high=high, low=low ) if exit_action is None: morning.legs = [{ "action": morning.action, "quantity": morning.quantity, "price": morning.price }] return morning legs = [{"action": morning.action, "quantity": morning.quantity, "price": morning.price}, { "action": exit_action.action, "quantity": exit_action.quantity, "price": exit_action.price }] composite_action = ( f"{morning.action}_THEN_{exit_action.action}" if morning.action not in {"HOLD", "DO_NOTHING"} else exit_action.action ) return SimAction( action_id=str(uuid.uuid4()), recommendation_id=rec.recommendation_id, portfolio_id=self.portfolio_id, symbol=rec.symbol, action=composite_action, quantity=exit_action.quantity, price=exit_action.price, cash_before=cash_before, cash_after=self.cash, reason=( f"session lifecycle: {morning.action} at {morning.price:.4f}; " f"{exit_action.action} at {exit_action.price:.4f}" ), rules=[*morning.rules, *exit_action.rules], created_at=datetime.now(timezone.utc), portfolio_before=before, portfolio_after=self.to_dict(), legs=legs, )