real-time portfolio Greeks: aggregating delta, gamma, theta, vega at scale

import asyncio
import redis.asyncio as aioredis
import pandas as pd
import numpy as np
from ib_insync import IB, Option, Contract
from typing import Optional
import logging
import json
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

REDIS_KEY_PORTFOLIO_GREEKS = "portfolio:greeks:current"
REDIS_KEY_POSITIONS = "portfolio:positions:raw"
REDIS_TTL_SECONDS = 30  # refresh every 30s


class PortfolioGreeksAggregator:
    """
    Pulls live position Greeks from IBKR and aggregates to portfolio level.
    Writes rollup to Redis for consumption by dashboard + risk alerts.
    """

    def __init__(self, ib: IB, redis_url: str = "redis://localhost:6379"):
        self.ib = ib
        self.redis_url = redis_url
        self._redis: Optional[aioredis.Redis] = None
        self._last_greeks: dict = {}

    async def connect_redis(self):
        self._redis = await aioredis.from_url(self.redis_url, decode_responses=True)

    async def fetch_positions_with_greeks(self) -> list[dict]:
        """Get all open positions and their live Greeks from IBKR."""
        positions = self.ib.positions()
        results = []

        for pos in positions:
            if pos.position == 0:
                continue

            contract = pos.contract
            if contract.secType not in ("OPT", "FUT"):
                continue

            # request market data with Greek calculation
            ticker = self.ib.reqMktData(contract, "106", False, False)
            await asyncio.sleep(0.1)  # give IBKR a moment

            greeks = ticker.modelGreeks
            if greeks is None:
                logger.warning(f"No Greeks for {contract.localSymbol}")
                continue

            multiplier = float(contract.multiplier or 100)

            results.append({
                "symbol": contract.localSymbol,
                "sec_type": contract.secType,
                "right": getattr(contract, "right", None),
                "position": pos.position,
                "multiplier": multiplier,
                "delta": (greeks.delta or 0.0) * pos.position * multiplier,
                "gamma": (greeks.gamma or 0.0) * pos.position * multiplier,
                "theta": (greeks.theta or 0.0) * pos.position * multiplier,
                "vega": (greeks.vega or 0.0) * pos.position * multiplier,
                "iv": greeks.impliedVol or 0.0,
                "underlying_price": greeks.undPrice or 0.0,
                "ts": datetime.now(timezone.utc).isoformat(),
            })

            self.ib.cancelMktData(contract)

        return results

    async def compute_portfolio_rollup(self, positions: list[dict]) -> dict:
        """Sum Greeks across all positions."""
        if not positions:
            return {}

        df = pd.DataFrame(positions)

        rollup = {
            "delta": round(df["delta"].sum(), 4),
            "gamma": round(df["gamma"].sum(), 6),
            "theta": round(df["theta"].sum(), 2),
            "vega": round(df["vega"].sum(), 2),
            "position_count": len(df),
            "avg_iv": round(df["iv"].mean(), 4),
            "ts": datetime.now(timezone.utc).isoformat(),
        }

        # strategy-level breakdown for dashboard
        rollup["by_symbol_prefix"] = (
            df.assign(prefix=df["symbol"].str[:3])
            .groupby("prefix")[["delta", "gamma", "theta", "vega"]]
            .sum()
            .round(4)
            .to_dict("index")
        )

        return rollup

    async def write_to_redis(self, rollup: dict, positions: list[dict]):
        """Persist current Greeks to Redis with TTL."""
        await self._redis.setex(
            REDIS_KEY_PORTFOLIO_GREEKS,
            REDIS_TTL_SECONDS,
            json.dumps(rollup),
        )
        await self._redis.setex(
            REDIS_KEY_POSITIONS,
            REDIS_TTL_SECONDS,
            json.dumps(positions),
        )

    async def run_loop(self, interval: float = 15.0):
        """Main update loop — runs every N seconds."""
        await self.connect_redis()

        logger.info(f"Starting Greeks aggregation loop (interval={interval}s)")
        while True:
            try:
                positions = await self.fetch_positions_with_greeks()
                rollup = await self.compute_portfolio_rollup(positions)

                if rollup:
                    await self.write_to_redis(rollup, positions)
                    self._last_greeks = rollup
                    logger.info(
                        f"Greeks updated | Δ={rollup['delta']:.2f} "
                        f"Γ={rollup['gamma']:.5f} "
                        f"Θ={rollup['theta']:.2f} "
                        f"V={rollup['vega']:.2f}"
                    )

            except Exception as e:
                logger.error(f"Greeks aggregation error: {e}", exc_info=True)

            await asyncio.sleep(interval)

import asyncpg
from datetime import datetime, timezone

CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS portfolio_greeks_history (
    ts TIMESTAMPTZ NOT NULL,
    delta DOUBLE PRECISION,
    gamma DOUBLE PRECISION,
    theta DOUBLE PRECISION,
    vega DOUBLE PRECISION,
    position_count INTEGER,
    avg_iv DOUBLE PRECISION
);
SELECT create_hypertable('portfolio_greeks_history', 'ts', if_not_exists => TRUE);
"""

async def persist_greeks_snapshot(pool: asyncpg.Pool, rollup: dict):
    async with pool.acquire() as conn:
        await conn.execute(
            """
            INSERT INTO portfolio_greeks_history
              (ts, delta, gamma, theta, vega, position_count, avg_iv)
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            """,
            datetime.fromisoformat(rollup["ts"]),
            rollup["delta"],
            rollup["gamma"],
            rollup["theta"],
            rollup["vega"],
            rollup["position_count"],
            rollup["avg_iv"],
        )