From c67eb82bba9f2276dcf4754df8a4d07fd8cac2a8 Mon Sep 17 00:00:00 2001 From: Jarno Date: Thu, 6 Aug 2026 18:55:59 +0300 Subject: [PATCH] Trade logic slop generated by copilot x( --- src/trading_bot/models/trade.py | 299 ++++++++++++++++++++++++++++++++ tests/test_trade.py | 62 +++++++ 2 files changed, 361 insertions(+) create mode 100644 src/trading_bot/models/trade.py create mode 100644 tests/test_trade.py diff --git a/src/trading_bot/models/trade.py b/src/trading_bot/models/trade.py new file mode 100644 index 0000000..1617a5b --- /dev/null +++ b/src/trading_bot/models/trade.py @@ -0,0 +1,299 @@ +"""Portfolio and open order inspection for Alpaca trading.""" + +from __future__ import annotations + +import argparse +import os +from dataclasses import dataclass +from typing import Any + +from alpaca.trading.client import TradingClient +from alpaca.trading.models import Position +from alpaca.trading.requests import GetOrdersRequest +from alpaca.trading.enums import QueryOrderStatus +from dotenv import load_dotenv + +load_dotenv() + +DEFAULT_SYMBOL = "SPY" + + +@dataclass(frozen=True) +class AlpacaOrderSummary: + symbol: str + side: str + qty: float + filled_qty: float + unfilled_qty: float + limit_price: float | None + order_type: str + status: str + + +@dataclass(frozen=True) +class AlpacaPortfolioSummary: + cash: float + spy_quantity: float + spy_market_value: float + spy_avg_entry_price: float | None + current_cash_to_spy_ratio: float | None + projected_cash_to_spy_ratio: float | None + open_spy_order_count: int + estimated_spy_price: float | None + + +def _string_to_float(value: Any, default: float = 0.0) -> float: + if value is None: + return default + if isinstance(value, (float, int)): + return float(value) + if isinstance(value, str): + value = value.strip() + if value == "": + return default + try: + return float(value) + except (TypeError, ValueError): + 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, + paper: bool = True, +) -> TradingClient: + if api_key is None or secret_key is None: + raise ValueError( + "Alpaca API key and secret key must be provided via CLI arguments or environment variables." + ) + + 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) + if value is None: + return None + price = _string_to_float(value, default=None) + 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) + ) + 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() + + return AlpacaOrderSummary( + symbol=symbol, + side=side, + qty=qty, + filled_qty=filled_qty, + unfilled_qty=max(qty - filled_qty, 0.0), + limit_price=limit_price, + order_type=order_type, + status=status, + ) + + +def _find_spy_position(positions: list[Any], 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() + 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, + ) + return quantity, market_value, avg_entry_price + + return 0.0, 0.0, None + + +def _safe_ratio(numerator: float, denominator: float) -> float | None: + if denominator <= 0: + return None + return numerator / denominator + + +def summarize_alpaca_portfolio( + account: Any, + positions: list[Position], + open_orders: list[Any], + symbol: str = DEFAULT_SYMBOL, +) -> AlpacaPortfolioSummary: + cash = _string_to_float(_get_attribute(account, "cash", default=0.0)) + spy_quantity, spy_market_value, spy_avg_entry_price = _find_spy_position( + positions, symbol=symbol + ) + current_ratio = _safe_ratio(cash, spy_market_value) + + spy_open_orders = [ + _build_order_summary(order) + for order in open_orders + if _build_order_summary(order).symbol == symbol.upper() + ] + + net_open_spy_qty = 0.0 + open_order_price = None + for order_summary in spy_open_orders: + if order_summary.side == "sell": + net_open_spy_qty -= order_summary.unfilled_qty + else: + net_open_spy_qty += order_summary.unfilled_qty + if open_order_price is None and order_summary.limit_price is not None: + open_order_price = order_summary.limit_price + + estimated_price = None + if spy_quantity > 0 and spy_market_value > 0: + estimated_price = spy_market_value / spy_quantity + elif open_order_price is not None: + estimated_price = open_order_price + + projected_cash = None + projected_spy_value = None + projected_ratio = None + if estimated_price is not None: + projected_cash = cash - (net_open_spy_qty * estimated_price) + projected_spy_value = spy_market_value + (net_open_spy_qty * estimated_price) + projected_ratio = _safe_ratio(projected_cash, projected_spy_value) + + return AlpacaPortfolioSummary( + cash=cash, + spy_quantity=spy_quantity, + spy_market_value=spy_market_value, + spy_avg_entry_price=spy_avg_entry_price, + current_cash_to_spy_ratio=current_ratio, + projected_cash_to_spy_ratio=projected_ratio, + open_spy_order_count=len(spy_open_orders), + estimated_spy_price=estimated_price, + ) + + +def fetch_alpaca_portfolio_summary( + api_key: str | None = None, + secret_key: str | None = None, + paper: bool = True, + symbol: str = DEFAULT_SYMBOL, +) -> AlpacaPortfolioSummary: + client = create_alpaca_trading_client( + api_key=api_key, + secret_key=secret_key, + paper=paper, + ) + account = client.get_account() + positions = client.get_all_positions() + if not isinstance(positions, list): + raise TypeError(f"Expected positions to be a list but got: {type(positions)}") + orders = client.get_orders(GetOrdersRequest(status=QueryOrderStatus.OPEN)) + if not isinstance(orders, list): + raise TypeError(f"Expected orders to be a list but got: {type(orders)}") + return summarize_alpaca_portfolio(account, positions, orders, symbol=symbol) + + +def print_alpaca_portfolio_summary(summary: AlpacaPortfolioSummary) -> None: + print("Alpaca portfolio summary") + print(f" cash: ${summary.cash:,.2f}") + print(f" SPY quantity: {summary.spy_quantity:.4f}") + print(f" SPY market value: ${summary.spy_market_value:,.2f}") + if summary.spy_avg_entry_price is not None: + print(f" SPY avg entry price: ${summary.spy_avg_entry_price:,.4f}") + if summary.estimated_spy_price is not None: + print(f" estimated SPY price: ${summary.estimated_spy_price:,.4f}") + print( + f" current cash / SPY value ratio: " + + ( + f"{summary.current_cash_to_spy_ratio:.4f}" + if summary.current_cash_to_spy_ratio is not None + else "n/a" + ) + ) + print(f" open SPY order count: {summary.open_spy_order_count}") + print( + f" projected cash / SPY ratio after open SPY orders: " + + ( + f"{summary.projected_cash_to_spy_ratio:.4f}" + if summary.projected_cash_to_spy_ratio is not None + else "n/a" + ) + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Fetch Alpaca account cash, SPY position, and open SPY order details." + ) + parser.add_argument( + "--api-key", + default=os.getenv("ALPACA_API_KEY"), + help="Alpaca API Key ID (defaults to ALPACA_API_KEY env var).", + ) + parser.add_argument( + "--secret-key", + default=os.getenv("ALPACA_SECRET_KEY"), + help="Alpaca Secret Key (defaults to ALPACA_SECRET_KEY env var).", + ) + parser.add_argument( + "--symbol", + default=DEFAULT_SYMBOL, + help="Symbol to inspect for SPY exposure. Defaults to SPY.", + ) + parser.add_argument( + "--paper", + action="store_true", + help="Force Alpaca paper trading mode. Defaults to paper if no base URL is configured.", + ) + return parser.parse_args() + + +def main() -> None: + args = parse_args() + summary = fetch_alpaca_portfolio_summary( + api_key=args.api_key, + secret_key=args.secret_key, + paper=args.paper if args.paper else True, + symbol=args.symbol, + ) + print_alpaca_portfolio_summary(summary) + + +if __name__ == "__main__": + main() diff --git a/tests/test_trade.py b/tests/test_trade.py new file mode 100644 index 0000000..f408805 --- /dev/null +++ b/tests/test_trade.py @@ -0,0 +1,62 @@ +from trading_bot.models.trade import summarize_alpaca_portfolio + + +def test_calculate_cash_to_spy_ratio() -> None: + account = {"cash": "1000.00"} + positions = [{"symbol": "SPY", "qty": "10", "market_value": "5000", "avg_entry_price": "500"}] + orders = [] + + summary = summarize_alpaca_portfolio(account, positions, orders) + + assert summary.cash == 1000.0 + assert summary.spy_quantity == 10.0 + assert summary.spy_market_value == 5000.0 + assert summary.current_cash_to_spy_ratio == 0.2 + assert summary.projected_cash_to_spy_ratio == 0.2 + assert summary.open_spy_order_count == 0 + + +def test_projected_ratio_includes_open_spy_buy_order() -> None: + account = {"cash": "1000.00"} + positions = [{"symbol": "SPY", "qty": "10", "market_value": "5000", "avg_entry_price": "500"}] + orders = [ + { + "symbol": "SPY", + "side": "buy", + "qty": "2", + "filled_qty": "0", + "limit_price": "500", + "status": "open", + "type": "limit", + } + ] + + summary = summarize_alpaca_portfolio(account, positions, orders) + + assert summary.open_spy_order_count == 1 + assert summary.estimated_spy_price == 500.0 + assert summary.projected_cash_to_spy_ratio == 0.14285714285714285 + + +def test_projected_ratio_handles_no_spy_position_but_open_order() -> None: + account = {"cash": "1500.00"} + positions = [] + orders = [ + { + "symbol": "SPY", + "side": "buy", + "qty": "3", + "filled_qty": "0", + "limit_price": "500", + "status": "open", + "type": "limit", + } + ] + + summary = summarize_alpaca_portfolio(account, positions, orders) + + assert summary.spy_quantity == 0.0 + assert summary.spy_market_value == 0.0 + assert summary.current_cash_to_spy_ratio is None + assert summary.projected_cash_to_spy_ratio == 1.0 + assert summary.open_spy_order_count == 1