diff --git a/src/trading_bot/models/trade.py b/src/trading_bot/models/trade.py index 1617a5b..ae9c0b4 100644 --- a/src/trading_bot/models/trade.py +++ b/src/trading_bot/models/trade.py @@ -8,9 +8,9 @@ from dataclasses import dataclass from typing import Any from alpaca.trading.client import TradingClient -from alpaca.trading.models import Position +from alpaca.trading.models import Order, Position, TradeAccount from alpaca.trading.requests import GetOrdersRequest -from alpaca.trading.enums import QueryOrderStatus +from alpaca.trading.enums import OrderType, QueryOrderStatus from dotenv import load_dotenv load_dotenv() @@ -57,21 +57,6 @@ def _string_to_float(value: Any, default: float = 0.0) -> float: return default -def _get_attribute(obj: Any, *names: str, default: Any = None) -> Any: - if obj is None: - return default - if isinstance(obj, dict): - for name in names: - if name in obj: - return obj[name] - return default - - for name in names: - if hasattr(obj, name): - return getattr(obj, name) - return default - - def create_alpaca_trading_client( api_key: str | None = None, secret_key: str | None = None, @@ -85,34 +70,22 @@ def create_alpaca_trading_client( return TradingClient(api_key, secret_key, paper=paper) -def _extract_unfilled_qty(order: Any) -> float: - qty = _string_to_float( - _get_attribute(order, "qty", "quantity", "order_qty", default=0) - ) - filled = _string_to_float( - _get_attribute(order, "filled_qty", "filled_quantity", default=0) - ) - return max(qty - filled, 0.0) - - -def _extract_order_limit_price(order: Any) -> float | None: - value = _get_attribute(order, "limit_price", "limitPrice", default=None) +def _extract_order_limit_price(order: Order) -> float | None: + value = order.limit_price if value is None: return None - price = _string_to_float(value, default=None) + price = _string_to_float(value, default=0.0) return price if price is not None and price > 0 else None -def _build_order_summary(order: Any) -> AlpacaOrderSummary: - symbol = str(_get_attribute(order, "symbol", "asset_symbol", default=DEFAULT_SYMBOL)).upper() - side = str(_get_attribute(order, "side", default="buy")).lower() - qty = _string_to_float(_get_attribute(order, "qty", "quantity", default=0)) - filled_qty = _string_to_float( - _get_attribute(order, "filled_qty", "filled_quantity", default=0) - ) +def _build_order_summary(order: Order) -> AlpacaOrderSummary: + symbol = str(order.symbol).upper() + side = str(order.side).lower() + qty = _string_to_float(order.qty or "0.0") + filled_qty = _string_to_float(order.filled_qty or "0.0") limit_price = _extract_order_limit_price(order) - order_type = str(_get_attribute(order, "type", default="unknown")) - status = str(_get_attribute(order, "status", default="unknown")).lower() + order_type = order.type or OrderType.MARKET + status = order.status or QueryOrderStatus.OPEN return AlpacaOrderSummary( symbol=symbol, @@ -126,23 +99,16 @@ def _build_order_summary(order: Any) -> AlpacaOrderSummary: ) -def _find_spy_position(positions: list[Any], symbol: str = DEFAULT_SYMBOL) -> tuple[float, float, float | None]: +def _find_spy_position(positions: list[Position], symbol: str = DEFAULT_SYMBOL) -> tuple[float, float, float | None]: normalized = symbol.upper() for position in positions: - position_symbol = str( - _get_attribute(position, "symbol", "asset_symbol", default="") - ).upper() + position_symbol = position.symbol.upper() if position_symbol != normalized: continue - quantity = _string_to_float(_get_attribute(position, "qty", "quantity", default=0)) - market_value = _string_to_float( - _get_attribute(position, "market_value", "marketValue", default=0) - ) - avg_entry_price = _string_to_float( - _get_attribute(position, "avg_entry_price", "avgEntryPrice", default=None), - default=None, - ) + quantity = _string_to_float(position.qty or "0.0") + market_value = _string_to_float(position.market_value or "0.0") + avg_entry_price = _string_to_float(position.avg_entry_price) return quantity, market_value, avg_entry_price return 0.0, 0.0, None @@ -155,12 +121,12 @@ def _safe_ratio(numerator: float, denominator: float) -> float | None: def summarize_alpaca_portfolio( - account: Any, + account: TradeAccount, positions: list[Position], - open_orders: list[Any], + open_orders: list[Order], symbol: str = DEFAULT_SYMBOL, ) -> AlpacaPortfolioSummary: - cash = _string_to_float(_get_attribute(account, "cash", default=0.0)) + cash = _string_to_float(account.cash or "0.0") spy_quantity, spy_market_value, spy_avg_entry_price = _find_spy_position( positions, symbol=symbol ) @@ -220,6 +186,8 @@ def fetch_alpaca_portfolio_summary( paper=paper, ) account = client.get_account() + if not isinstance(account, TradeAccount): + raise TypeError(f"Expected account to be a TradeAccount but got: {type(account)}") positions = client.get_all_positions() if not isinstance(positions, list): raise TypeError(f"Expected positions to be a list but got: {type(positions)}")