Trade logic slop generated by copilot x(

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()

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