scanner/factor_sources.py

"""Data-source interface for the four "real" institutional flow factors.

The scanner runs on Hugging Face Spaces (no live market data feed) but the
math should be identical whether the data comes from a local Futu OpenD
gateway or from the bundled synthetic stubs.  This module defines a
:func:`get_data_source` factory that returns either:

* :class:`StubDataSource`  - reads pre-baked JSON / parquet files in
  ``data/stubs/`` (default on the Space, no network required)
* :class:`FutuDataSource`  - live Level-2 / options / tick data via the
  Futu OpenD gateway (set ``FSCANNER_DATA_SOURCE=futu``)

Each factor module (``l2_factor``, ``options_factor``, ``tick_factor``,
``intraday_factor``) calls the relevant method on whichever source is
active.
"""

from __future__ import annotations

import json
import os
import random
from datetime import datetime, timedelta
from typing import Optional, Protocol

import pandas as pd

from . import paths


# ---------------------------------------------------------------------------
# Interface
# ---------------------------------------------------------------------------

class FactorDataSource(Protocol):
    """Abstract data source for the four institutional-flow factors."""

    name: str

    def get_l2_snapshot(self, ticker: str) -> Optional[dict]:
        """Return the most recent Level-2 order-book snapshot for ``ticker``.

        Schema::

            {
                "ticker": "AAPL",
                "ts": "2026-06-02T14:30:00Z",
                "bids": [[price, size, mpid, age_sec], ...],   # top N
                "asks": [[price, size, mpid, age_sec], ...],
            }

        ``age_sec`` is how long the order has been sitting on the book;
        it is used to mitigate spoofing.
        """
        ...

    def get_options_history(
        self, ticker: str, lookback_days: int = 20
    ) -> Optional[pd.DataFrame]:
        """Return daily options-chain aggregates for ``ticker``.

        Schema (one row per ``(date, kind, moneyness_bucket)``)::

            date        datetime64
            kind        'call' | 'put'
            moneyness   'itm' | 'atm' | 'otm'  (delta-based bucket)
            volume      int
            oi          int
            avg_iv      float
        """
        ...

    def get_ticks(
        self, ticker: str, date: Optional[str] = None
    ) -> Optional[pd.DataFrame]:
        """Return raw trade ticks for ``ticker`` on ``date`` (YYYY-MM-DD).

        Schema::

            ts        datetime64
            price     float
            size      int
            bid       float   # NBBO bid at time of trade
            ask       float   # NBBO ask at time of trade
        """
        ...

    def get_intraday_bars(
        self, ticker: str, date: Optional[str] = None, bar_minutes: int = 5
    ) -> Optional[pd.DataFrame]:
        """Return pre-aggregated intraday bars (5-min default).

        Schema::

            bar_start  datetime64
            open        float
            high        float
            low         float
            close       float
            volume      int
            buy_vol     int   # buy-initiated
            sell_vol    int   # sell-initiated
        """
        ...


# ---------------------------------------------------------------------------
# Stub data source (bundled synthetic data)
# ---------------------------------------------------------------------------

class StubDataSource:
    """Reads pre-baked stub data from ``data/stubs/``.

    The stub data is generated at module-build time by
    ``data/stubs/_build_stubs.py`` and committed to the repo so the app
    can demonstrate the four factors on the Space without any live feed.
    """

    name = "stub"

    def __init__(self, stub_dir: Optional[str] = None) -> None:
        self.dir = stub_dir or paths.STUB_DIR

    def get_l2_snapshot(self, ticker: str) -> Optional[dict]:
        path = os.path.join(self.dir, "l2", f"{ticker}.json")
        if not os.path.exists(path):
            return self._synth_l2(ticker)
        try:
            with open(path, "r", encoding="utf-8") as fh:
                return json.load(fh)
        except Exception:
            return None

    def get_options_history(
        self, ticker: str, lookback_days: int = 20
    ) -> Optional[pd.DataFrame]:
        path = os.path.join(self.dir, "options", f"{ticker}.parquet")
        if not os.path.exists(path):
            return self._synth_options(ticker, lookback_days)
        try:
            df = pd.read_parquet(path)
            if df.empty:
                return None
            cutoff = pd.Timestamp.utcnow().tz_localize(None) - pd.Timedelta(days=lookback_days)
            return df[df["date"] >= cutoff].reset_index(drop=True)
        except Exception:
            return None

    def get_ticks(
        self, ticker: str, date: Optional[str] = None
    ) -> Optional[pd.DataFrame]:
        path = os.path.join(self.dir, "ticks", f"{ticker}.parquet")
        if not os.path.exists(path):
            return self._synth_ticks(ticker, date)
        try:
            return pd.read_parquet(path)
        except Exception:
            return None

    def get_intraday_bars(
        self, ticker: str, date: Optional[str] = None, bar_minutes: int = 5
    ) -> Optional[pd.DataFrame]:
        path = os.path.join(self.dir, "intraday", f"{ticker}.parquet")
        if not os.path.exists(path):
            return self._synth_intraday(ticker)
        try:
            return pd.read_parquet(path)
        except Exception:
            return None

    # -- Synthetic fallbacks (deterministic per ticker) ------------------

    @staticmethod
    def _seeded(ticker: str) -> random.Random:
        return random.Random(f"stub-{ticker}")

    def _synth_l2(self, ticker: str) -> dict:
        rng = self._seeded(ticker)
        mid = rng.uniform(20, 500)
        spread = mid * 0.0005
        bids, asks = [], []
        for i in range(10):
            bp = mid - spread / 2 - i * spread * 0.5
            ap = mid + spread / 2 + i * spread * 0.5
            bs = int(rng.lognormvariate(6, 1.2))
            as_ = int(rng.lognormvariate(6, 1.2))
            bids.append([round(bp, 2), bs, "NSDQ", round(rng.uniform(1.2, 30), 1)])
            asks.append([round(ap, 2), as_, "NSDQ", round(rng.uniform(1.2, 30), 1)])
        return {
            "ticker": ticker,
            "ts": datetime.utcnow().isoformat() + "Z",
            "bids": bids,
            "asks": asks,
        }

    def _synth_options(
        self, ticker: str, lookback_days: int
    ) -> pd.DataFrame:
        rng = self._seeded(ticker + "-opt")
        today = pd.Timestamp.utcnow().tz_localize(None).normalize()
        rows = []
        for d in range(lookback_days):
            date = today - pd.Timedelta(days=d)
            for kind in ("call", "put"):
                for bucket in ("itm", "atm", "otm"):
                    base = rng.lognormvariate(7, 0.8)
                    rows.append({
                        "date": date,
                        "kind": kind,
                        "moneyness": bucket,
                        "volume": int(base * rng.uniform(0.5, 1.5)),
                        "oi": int(base * rng.uniform(3, 10)),
                        "avg_iv": rng.uniform(0.18, 0.65),
                    })
        return pd.DataFrame(rows)

    def _synth_ticks(self, ticker: str, date: Optional[str]) -> pd.DataFrame:
        rng = self._seeded(ticker + "-ticks")
        n = rng.randint(800, 1500)
        base = rng.uniform(20, 500)
        ts0 = pd.Timestamp(date or "2026-06-02", tz=None) + pd.Timedelta(hours=9, minutes=30)
        ticks = []
        price = base
        for i in range(n):
            dt = pd.Timedelta(seconds=i * 1.5 + rng.uniform(0, 1.5))
            price *= 1 + rng.gauss(0, 0.0005)
            spread = price * 0.0003
            side = rng.random()
            sz = int(rng.choices([50, 100, 200, 500, 1000, 5000, 10000, 20000],
                                  weights=[0.25, 0.25, 0.15, 0.15, 0.10, 0.05, 0.03, 0.02])[0])
            ticks.append({
                "ts": ts0 + dt,
                "price": round(price, 4),
                "size": sz,
                "bid": round(price - spread / 2, 4),
                "ask": round(price + spread / 2, 4),
            })
        return pd.DataFrame(ticks)

    def _synth_intraday(self, ticker: str) -> pd.DataFrame:
        rng = self._seeded(ticker + "-intra")
        bars = []
        day = pd.Timestamp("2026-06-02") + pd.Timedelta(hours=9, minutes=30)
        price = rng.uniform(20, 500)
        for i in range(78):  # 78 * 5min = 6.5h trading day
            ts = day + pd.Timedelta(minutes=i * 5)
            o = price
            ret = rng.gauss(0, 0.003)
            c = o * (1 + ret)
            h = max(o, c) * (1 + abs(rng.gauss(0, 0.0015)))
            l = min(o, c) * (1 - abs(rng.gauss(0, 0.0015)))
            v = int(rng.lognormvariate(13, 0.6))
            buy_ratio = 0.5 + rng.gauss(0, 0.08)
            buy_ratio = max(0.30, min(0.70, buy_ratio))
            bv = int(v * buy_ratio)
            bars.append({
                "bar_start": ts,
                "open": o, "high": h, "low": l, "close": c, "volume": v,
                "buy_vol": bv, "sell_vol": v - bv,
            })
            price = c
        return pd.DataFrame(bars)


# ---------------------------------------------------------------------------
# Live Futu OpenD adapter (optional)
# ---------------------------------------------------------------------------

class FutuDataSource:
    """Live Level-2 / options / tick data via Futu OpenD.

    To use this, the user runs Futu OpenD locally and sets::

        export FSCANNER_DATA_SOURCE=futu
        export FUTU_OPEND_HOST=127.0.0.1
        export FUTU_OPEND_PORT=11111

    The Space will not have OpenD reachable, so :func:`get_data_source`
    will fall back to the stub source automatically.
    """

    name = "futu"

    def __init__(self, host: str = "127.0.0.1", port: int = 11111) -> None:
        self.host = host
        self.port = port
        self._ctx = None

    def _ensure_ctx(self):
        if self._ctx is None:
            try:
                import futu as ft  # type: ignore
            except ImportError as e:
                raise RuntimeError(
                    "futu-api is not installed. `pip install futu-api` and "
                    "make sure Futu OpenD is running."
                ) from e
            self._ctx = ft.OpenQuoteContext(host=self.host, port=self.port)
        return self._ctx

    def get_l2_snapshot(self, ticker: str) -> Optional[dict]:
        try:
            ctx = self._ensure_ctx()
            code = f"US.{ticker}"
            ret, data = ctx.get_order_book(code, num=10)
            if ret != 0 or data is None or data.empty:
                return None
            # Futu returns bids and asks as two DataFrames
            bids_df = data[0]  # (price, volume, turnover, orderid)
            asks_df = data[1]
            return {
                "ticker": ticker,
                "ts": datetime.utcnow().isoformat() + "Z",
                "bids": bids_df.values.tolist(),
                "asks": asks_df.values.tolist(),
            }
        except Exception:
            return None

    def get_options_history(
        self, ticker: str, lookback_days: int = 20
    ) -> Optional[pd.DataFrame]:
        # Simplified: would need to pull chain + historical IV.  Stub fallback
        # for the demo.
        return StubDataSource().get_options_history(ticker, lookback_days)

    def get_ticks(
        self, ticker: str, date: Optional[str] = None
    ) -> Optional[pd.DataFrame]:
        # Live tick feeds are behind a paid tier; stub fallback.
        return StubDataSource().get_ticks(ticker, date)

    def get_intraday_bars(
        self, ticker: str, date: Optional[str] = None, bar_minutes: int = 5
    ) -> Optional[pd.DataFrame]:
        return StubDataSource().get_intraday_bars(ticker, date, bar_minutes)


# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------

_source: Optional[FactorDataSource] = None


def get_data_source() -> FactorDataSource:
    """Return the configured data source (singleton)."""
    global _source
    if _source is not None:
        return _source
    which = os.environ.get("FSCANNER_DATA_SOURCE", "stub").lower()
    if which == "futu":
        host = os.environ.get("FUTU_OPEND_HOST", "127.0.0.1")
        port = int(os.environ.get("FUTU_OPEND_PORT", "11111"))
        try:
            _source = FutuDataSource(host=host, port=port)
        except Exception:
            _source = StubDataSource()
    else:
        _source = StubDataSource()
    return _source


def reset_data_source() -> None:
    """For tests - force a re-init on next :func:`get_data_source` call."""
    global _source
    _source = None