Generally quite nicely working already

backend/flex_lots.py

@@ -0,0 +1,144 @@
+"""Parse lot-level open positions from IB Flex XML files."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+import xml.etree.ElementTree as ET
+
+
+def _as_float(value, default=None):
+    try:
+        if value in (None, ""):
+            return default
+        return float(value)
+    except Exception:
+        return default
+
+
+def _normalize_expiry(expiry: str | None) -> str | None:
+    expiry = (expiry or "").strip()
+    if len(expiry) == 8 and expiry.isdigit():
+        return f"{expiry[0:4]}-{expiry[4:6]}-{expiry[6:8]}"
+    return expiry or None
+
+
+def _normalize_buy_time(open_dt: str | None) -> str | None:
+    open_dt = (open_dt or "").strip()
+    if not open_dt or ";" not in open_dt:
+        return open_dt or None
+    date_part, time_part = open_dt.split(";", 1)
+    if len(date_part) == 8 and len(time_part) == 6:
+        return f"{date_part[0:4]}-{date_part[4:6]}-{date_part[6:8]} {time_part[0:2]}:{time_part[2:4]}:{time_part[4:6]}"
+    return open_dt
+
+
+def _compute_cagr_percent(cost_basis, gain_dollars, buy_time: str | None):
+    if cost_basis in (None, 0) or gain_dollars is None or not buy_time:
+        return None
+    try:
+        start = datetime.fromisoformat(buy_time).replace(tzinfo=timezone.utc)
+        now = datetime.now(timezone.utc)
+        years = max((now - start).total_seconds() / (365.25 * 24 * 3600), 1e-9)
+        start_value = abs(float(cost_basis))
+        end_value = start_value + float(gain_dollars)
+        if start_value <= 0 or end_value <= 0:
+            return None
+        return ((end_value / start_value) ** (1.0 / years) - 1.0) * 100.0
+    except Exception:
+        return None
+
+
+def _compute_days_since_buy(buy_time: str | None):
+    if not buy_time:
+        return None
+    try:
+        start = datetime.fromisoformat(buy_time).replace(tzinfo=timezone.utc)
+        now = datetime.now(timezone.utc)
+        return max((now - start).days, 0)
+    except Exception:
+        return None
+
+
+def _display_symbol(attrs: dict) -> str:
+    description = (attrs.get("description") or "").strip()
+    if description:
+        return description
+    return (attrs.get("underlyingSymbol") or attrs.get("symbol") or "").strip()
+
+
+def _instrument_key(attrs: dict) -> str:
+    conid = (attrs.get("conid") or "").strip()
+    if conid:
+        return f"CONID:{conid}"
+    return (attrs.get("symbol") or "").strip()
+
+
+def parse_flex_lots(xml_path: str | Path) -> list[dict]:
+    root = ET.parse(xml_path).getroot()
+    lots: list[dict] = []
+    for node in root.findall(".//OpenPosition"):
+        attrs = node.attrib
+        if (attrs.get("levelOfDetail") or "").upper() != "LOT":
+            continue
+
+        security_type = (attrs.get("assetCategory") or "").upper() or "STK"
+        multiplier = _as_float(attrs.get("multiplier"), 1.0) or 1.0
+        raw_position = _as_float(attrs.get("position"), 0.0) or 0.0
+        side = (attrs.get("side") or "").strip().lower()
+        qty = -abs(raw_position) if side == "short" or raw_position < 0 else abs(raw_position)
+
+        mark_price = _as_float(attrs.get("markPrice"))
+        market_price = mark_price
+        market_value = _as_float(attrs.get("positionValue"))
+        cost_basis = _as_float(attrs.get("costBasisMoney"))
+        avg_price = _as_float(attrs.get("costBasisPrice"))
+        if avg_price is None:
+            open_price = _as_float(attrs.get("openPrice"))
+            avg_price = open_price
+        if avg_price is None and qty:
+            divisor = qty * multiplier if security_type == "OPT" else qty
+            avg_price = abs(cost_basis / divisor) if cost_basis is not None and divisor else None
+        buy_time = _normalize_buy_time(attrs.get("openDateTime"))
+        gain_dollars = _as_float(attrs.get("fifoPnlUnrealized"))
+        gain_percent = None
+        if cost_basis not in (None, 0) and gain_dollars is not None:
+            gain_percent = (gain_dollars / abs(cost_basis)) * 100.0
+        cagr_percent = _compute_cagr_percent(cost_basis, gain_dollars, buy_time)
+        days_since_buy = _compute_days_since_buy(buy_time)
+        ltcg = "X" if days_since_buy is not None and days_since_buy >= 365 else ""
+
+        lots.append({
+            "instrument_key": _instrument_key(attrs),
+            "con_id": _as_float(attrs.get("conid")),
+            "symbol": attrs.get("symbol"),
+            "display_symbol": _display_symbol(attrs),
+            "security_type": security_type,
+            "underlying_symbol": attrs.get("underlyingSymbol") or attrs.get("symbol"),
+            "expiry": _normalize_expiry(attrs.get("expiry")),
+            "strike": _as_float(attrs.get("strike")),
+            "option_type": (attrs.get("putCall") or "").upper() or None,
+            "multiplier": str(int(multiplier)) if float(multiplier).is_integer() else str(multiplier),
+            "qty": qty,
+            "avg_price": avg_price,
+            "cost_basis": cost_basis,
+            "market_price": market_price,
+            "market_value": market_value,
+            "gain_dollars": gain_dollars,
+            "gain_percent": gain_percent,
+            "cagr_percent": cagr_percent,
+            "days_since_buy": days_since_buy,
+            "ltcg": ltcg,
+            "buy_time": buy_time,
+        })
+    return lots
+
+
+def load_latest_flex_lots(data_dir: str | Path) -> tuple[str | None, list[dict]]:
+    data_path = Path(data_dir)
+    if not data_path.exists():
+        return None, []
+    xml_files = sorted(data_path.glob("flex_*.xml"))
+    if not xml_files:
+        return None, []
+    latest = xml_files[-1]
+    return str(latest), parse_flex_lots(latest)