IB-Dashboard/backend/flex_lots.py

"""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, market_value, 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)
        gain_percent = (float(gain_dollars) / abs(float(cost_basis))) * 100.0
        if years < 1.0:
            return gain_percent
        start_value = abs(float(cost_basis))
        if float(cost_basis) < 0:
            end_liability = abs(float(market_value)) if market_value is not None else abs(float(cost_basis) + float(gain_dollars))
            if start_value <= 0 or end_liability <= 0:
                return None
            return (((start_value / end_liability) ** (1.0 / years)) - 1.0) * 100.0
        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 _signed_quantity(raw_position, side: str, cost_basis, market_value):
    quantity = abs(raw_position or 0.0)
    if quantity == 0:
        return 0.0
    normalized_side = (side or "").strip().lower()
    if normalized_side == "short" or (raw_position or 0.0) < 0:
        return -quantity
    if cost_basis is not None and float(cost_basis) < 0:
        return -quantity
    if market_value is not None and float(market_value) < 0:
        return -quantity
    return quantity


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
        fx_rate_to_base = _as_float(attrs.get("fxRateToBase"), 1.0) or 1.0
        raw_position = _as_float(attrs.get("position"), 0.0) or 0.0
        side = (attrs.get("side") or "").strip().lower()

        mark_price = _as_float(attrs.get("markPrice"))
        market_price = mark_price
        market_value = _as_float(attrs.get("positionValue"))
        cost_basis_base = _as_float(attrs.get("costBasisMoney"))
        capital_gain_base = _as_float(attrs.get("unrealizedCapitalGainsPnl"))
        gain_dollars = _as_float(attrs.get("fifoPnlUnrealized"))
        qty = _signed_quantity(raw_position, side, cost_basis_base, market_value)

        # Flex OpenPosition rows mix currencies for non-base holdings:
        # mark/positionValue are in instrument currency, while costBasis/openPrice
        # are in account base. Reconstruct local-cost values so they line up with TWS.
        cost_basis = cost_basis_base
        if market_value is not None and capital_gain_base is not None and fx_rate_to_base:
            cost_basis = market_value - (capital_gain_base / fx_rate_to_base)

        avg_price = None
        if qty:
            divisor = abs(qty) * multiplier if security_type == "OPT" else abs(qty)
            avg_price = abs(cost_basis / divisor) if cost_basis is not None and divisor else None
        if avg_price is None:
            avg_price = _as_float(attrs.get("costBasisPrice"))
        if avg_price is None:
            avg_price = _as_float(attrs.get("openPrice"))
        buy_time = _normalize_buy_time(attrs.get("openDateTime"))

        # Prefer local-currency capital gain to keep gain/cost/value in one currency.
        if capital_gain_base is not None and fx_rate_to_base:
            gain_dollars = capital_gain_base / fx_rate_to_base

        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, market_value, 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),
            "currency": attrs.get("currency") or None,
            "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)