OpenFin/backend/services/portfolio_risk.py

"""Simulation portfolio risk analysis — mirrors IBKR risk module for logged trades."""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from services.database import get_conn

_BEARISH_KEYWORDS = {"bear", "put", "short", "sell", "vente", "baissier"}

# Ticker → asset_class mapping for common instruments (fallback when pattern not in DB)
_TICKER_AC: Dict[str, str] = {
    # Energy
    "CL=F": "energy", "BZ=F": "energy", "NG=F": "energy", "RB=F": "energy",
    "HO=F": "energy", "USO": "energy", "XLE": "energy", "XOP": "energy", "OIL": "energy",
    # Metals
    "GC=F": "metals", "SI=F": "metals", "HG=F": "metals", "PL=F": "metals",
    "PA=F": "metals", "GLD": "metals", "SLV": "metals", "GDX": "metals", "GDXJ": "metals",
    # Agriculture
    "ZW=F": "agriculture", "ZC=F": "agriculture", "ZS=F": "agriculture",
    "CT=F": "agriculture", "KC=F": "agriculture", "SB=F": "agriculture", "CC=F": "agriculture",
    "WEAT": "agriculture", "CORN": "agriculture", "SOYB": "agriculture",
    # Equity indices
    "^GSPC": "indices", "^DJI": "indices", "^NDX": "indices", "^RUT": "indices",
    "^VIX": "indices", "^FTSE": "indices", "^GDAXI": "indices", "^FCHI": "indices",
    "^N225": "indices", "^HSI": "indices", "^NSEI": "indices", "^BSESN": "indices",
    "^STOXX50E": "indices", "^IBEX": "indices",
    "SPY": "indices", "QQQ": "indices", "IWM": "indices", "DIA": "indices",
    "VXX": "indices", "UVXY": "indices", "SVXY": "indices",
    # Forex
    "EURUSD=X": "forex", "GBPUSD=X": "forex", "USDJPY=X": "forex",
    "AUDUSD=X": "forex", "USDCAD=X": "forex", "USDCHF=X": "forex",
    "NZDUSD=X": "forex", "EURGBP=X": "forex", "EURJPY=X": "forex",
    "GBPJPY=X": "forex", "USDCNH=X": "forex",
    "FXE": "forex", "UUP": "forex", "FXB": "forex", "FXY": "forex",
    # Rates
    "ZB=F": "rates", "ZN=F": "rates", "ZF=F": "rates", "ZT=F": "rates",
    "TLT": "rates", "IEF": "rates", "SHY": "rates", "HYG": "rates",
    "LQD": "rates", "EMB": "rates",
}


def _infer_asset_class(ticker: str) -> str:
    """Derive asset_class from ticker when not stored in DB."""
    t = (ticker or "").upper().strip()
    if t in _TICKER_AC:
        return _TICKER_AC[t]
    if ":" in t:  # NSE:RELIANCE, BSE:TCS, etc.
        return "equities"
    if t.endswith("=X") and len(t) >= 7:  # forex pairs like EURUSD=X
        return "forex"
    if t.endswith("=F"):  # generic futures
        return "energy"  # most unknown futures are commodities
    if t.startswith("^"):  # index
        return "indices"
    if t.isalpha() and len(t) <= 5:  # short alpha = equity
        return "equities"
    return "unknown"


def _direction(strategy: str) -> str:
    s = (strategy or "").lower()
    return "bearish" if any(kw in s for kw in _BEARISH_KEYWORDS) else "bullish"


def get_open_simulation_trades() -> List[Dict[str, Any]]:
    """Open trades enriched with asset_class — via stored column, JOIN fallback, then ticker inference."""
    conn = get_conn()
    rows = conn.execute("""
        SELECT tep.*,
               COALESCE(tep.asset_class, cp.asset_class) AS asset_class
        FROM trade_entry_prices tep
        LEFT JOIN custom_patterns cp ON cp.id = tep.pattern_id
        WHERE (tep.status IS NULL OR tep.status = 'open')
        ORDER BY tep.entry_date DESC
    """).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        if not d.get("asset_class"):
            d["asset_class"] = _infer_asset_class(d.get("underlying", ""))
        result.append(d)
    return result


def analyze_simulation_portfolio() -> Dict[str, Any]:
    """Full risk breakdown of the open simulation portfolio."""
    trades = get_open_simulation_trades()
    if not trades:
        return {
            "open_count": 0, "conflicts": [], "concentration": {},
            "by_underlying": {}, "alerts": [], "direction_exposure": {},
        }

    total = len(trades)

    # Group by asset_class
    by_class: Dict[str, List] = {}
    for t in trades:
        ac = (t.get("asset_class") or "unknown").lower()
        by_class.setdefault(ac, []).append(t)

    concentration = {
        ac: {
            "count": len(items),
            "pct": round(len(items) / total * 100, 1),
            "tickers": sorted({t["underlying"] for t in items}),
            "bullish": sum(1 for t in items if _direction(t.get("strategy", "")) == "bullish"),
            "bearish": sum(1 for t in items if _direction(t.get("strategy", "")) == "bearish"),
        }
        for ac, items in by_class.items()
    }

    # Group by underlying
    by_underlying: Dict[str, List] = {}
    for t in trades:
        u = (t.get("underlying") or "").upper()
        if u:
            by_underlying.setdefault(u, []).append(t)

    # Detect directional conflicts
    conflicts = []
    for underlying, group in by_underlying.items():
        dirs = [_direction(t.get("strategy", "")) for t in group]
        if "bullish" in dirs and "bearish" in dirs:
            conflicts.append({
                "underlying": underlying,
                "trades": [
                    {
                        "id": t["id"],
                        "strategy": t.get("strategy", ""),
                        "direction": _direction(t.get("strategy", "")),
                        "entry_date": t.get("entry_date", ""),
                        "pattern_name": t.get("pattern_name", ""),
                        "score_at_entry": t.get("score_at_entry", 0),
                        "asset_class": (t.get("asset_class") or "").lower(),
                    }
                    for t in group
                ],
            })

    # Direction exposure summary per class
    direction_exposure = {
        ac: {
            "bullish": data["bullish"],
            "bearish": data["bearish"],
            "net": data["bullish"] - data["bearish"],
            "bias": "bullish" if data["bullish"] > data["bearish"]
                    else "bearish" if data["bearish"] > data["bullish"]
                    else "neutral",
        }
        for ac, data in concentration.items()
    }

    # Build alerts — sorted by severity
    alerts: List[Dict] = []
    for c in conflicts:
        alerts.append({
            "type": "conflict", "level": "danger",
            "underlying": c["underlying"],
            "message": f"Positions contradictoires sur {c['underlying']} — {len(c['trades'])} trades opposés",
        })
    for ac, data in concentration.items():
        if data["pct"] >= 35:
            alerts.append({
                "type": "concentration", "level": "warning",
                "asset_class": ac, "pct": data["pct"],
                "message": f"{ac} = {data['pct']}% du portefeuille simulé ({data['count']} trades)",
            })
    for underlying, group in by_underlying.items():
        if len(group) >= 3:
            alerts.append({
                "type": "overweight", "level": "warning",
                "underlying": underlying, "count": len(group),
                "message": f"{len(group)} trades sur {underlying} — sur-exposition",
            })
    alerts.sort(key=lambda a: {"danger": 0, "warning": 1}.get(a["level"], 2))

    return {
        "open_count": total,
        "conflicts": conflicts,
        "concentration": concentration,
        "by_underlying": {u: len(g) for u, g in by_underlying.items()},
        "direction_exposure": direction_exposure,
        "alerts": alerts,
    }


def check_new_trade(underlying: str, strategy: str, asset_class: str) -> Dict[str, Any]:
    """Pre-entry check: would this new trade create conflicts or concentration issues?"""
    open_trades = get_open_simulation_trades()
    total = len(open_trades)

    new_dir = _direction(strategy)
    underlying_upper = underlying.upper()
    warnings: List[Dict] = []

    same_underlying = [t for t in open_trades if (t.get("underlying") or "").upper() == underlying_upper]
    for t in same_underlying:
        existing_dir = _direction(t.get("strategy", ""))
        if existing_dir != new_dir:
            warnings.append({
                "type": "conflict", "level": "danger",
                "trade_id": t["id"], "existing_direction": existing_dir,
                "message": f"Conflit : trade #{t['id']} est {existing_dir} sur {underlying} ({t.get('strategy', '')})",
            })
        else:
            warnings.append({
                "type": "accumulation", "level": "info",
                "trade_id": t["id"],
                "message": f"Accumulation {new_dir} sur {underlying} (trade #{t['id']} déjà en position)",
            })

    # Forecast concentration after adding this trade
    total_after = total + 1
    ac_key = (asset_class or "").lower()
    current_ac = sum(
        1 for t in open_trades
        if (t.get("asset_class") or "").lower() == ac_key
    )
    future_pct = round((current_ac + 1) / total_after * 100, 1) if total_after > 0 else 100
    if future_pct >= 35 and ac_key:
        warnings.append({
            "type": "concentration", "level": "warning",
            "asset_class": ac_key, "future_pct": future_pct,
            "message": f"Ce trade porterait {ac_key} à {future_pct}% du portefeuille simulé",
        })

    return {
        "warnings": warnings,
        "ok": not any(w["level"] == "danger" for w in warnings),
        "underlying": underlying,
        "strategy": strategy,
        "direction": new_dir,
        "total_open_after": total_after,
    }


def build_monitor_context(risk: Dict[str, Any]) -> str:
    """Build a compact text summary of portfolio risk for GPT-4o."""
    lines = [
        f"PORTEFEUILLE SIMULÉ — {risk['open_count']} positions ouvertes",
        "",
        "RÉPARTITION PAR CLASSE D'ACTIF:",
    ]
    for ac, data in sorted(risk["concentration"].items(), key=lambda x: -x[1]["pct"]):
        exp = risk["direction_exposure"].get(ac, {})
        lines.append(
            f"  {ac}: {data['pct']}% ({data['count']} trades) — "
            f"bull={exp.get('bullish', 0)} bear={exp.get('bearish', 0)} — "
            f"tickers: {', '.join(data['tickers'][:5])}"
        )

    if risk["conflicts"]:
        lines += ["", "CONFLITS DIRECTIONNELS (DANGER):"]
        for c in risk["conflicts"]:
            trades_desc = "; ".join(
                f"#{t['id']} {t['direction']} ({t['strategy']}) depuis {t['entry_date']}"
                for t in c["trades"]
            )
            lines.append(f"  {c['underlying']}: {trades_desc}")

    if risk["alerts"]:
        lines += ["", "ALERTES:"]
        for a in risk["alerts"]:
            lines.append(f"  [{a['level'].upper()}] {a['message']}")

    return "\n".join(lines)