#pragma once // OptionChainService — DataHub Producer for `option:chain:*` and friends. // // Phase 1 scope: // - Owns the option chain assembly path. Reads instruments from // InstrumentService, fetches quotes via the broker's IBroker::get_quotes, // enriches each strike with OI/IV/Greeks, and publishes the assembled // OptionChain on `option:chain:::`. // - Greeks/IV are computed during chain assembly (see enrich_with_greeks): // an IV solver + per-leg greeks, throttled and cached. Brokers that supply // greeks in their quotes (e.g. Fyers) are used directly. // - WebSocket OI fan-out is wired in Phase 3; Phase 1 ships polled // refresh through DataHub. // // The service is a singleton so it can be registered from main.cpp and // looked up from any consumer screen. It DOES NOT spawn Python directly // (D1) — Greeks routing goes through PythonWorker. It is the sole hub // producer for the option:* topic family (D2). // // Not thread-safe in the API surface; all entry points must be called // from the UI thread. Internal QtConcurrent work is marshalled back via // QMetaObject::invokeMethod / QPointer guards. #include "datahub/Producer.h" #include "services/options/OptionChainTypes.h" #include #include #include #include #include #include #include namespace fincept::services::options { class OptionChainService : public QObject, public fincept::datahub::Producer { Q_OBJECT public: static OptionChainService& instance(); /// Idempotent — registers as Producer for the option:* topic families and /// installs default policies. Called from main.cpp after datahub /// metatype registration. void ensure_registered_with_hub(); // ── fincept::datahub::Producer ────────────────────────────────────────── QStringList topic_patterns() const override; void refresh(const QStringList& topics) override; int max_requests_per_sec() const override; // ── Consumer-facing helpers ───────────────────────────────────────────── /// One-shot read of the current cached chain, or std::nullopt when no /// snapshot exists. Triggers no fetch — wrapper around DataHub::peek. /// Streaming consumers should subscribe to the topic instead. QString chain_topic(const QString& broker_id, const QString& underlying, const QString& expiry) const; /// Last published chain snapshot — available immediately for sub-tabs /// that are lazily constructed after the initial chain publish. const fincept::services::options::OptionChain& last_chain() const { return last_chain_; } /// List underlyings + expiries from InstrumentService for picker UIs. /// For broker_id == "databento", returns hardcoded US equity list / cached expiries. QStringList list_underlyings(const QString& broker_id) const; QStringList list_expiries(const QString& broker_id, const QString& underlying) const; /// Async expiry fetch for Databento — calls Python script, caches result. void list_databento_expiries(const QString& underlying, std::function callback); /// Pin extra contract symbols into the live WS subscription window for a chain /// topic. Replaces any prior pin set for that topic; an empty list clears pins. /// Call from the main thread only. void pin_contracts(const QString& topic, const QStringList& symbols); /// Returns the currently-pinned symbols for `topic` (empty when no pins set). QStringList pinned_contracts_for(const QString& topic) const { return pinned_contracts_.value(topic); } signals: /// Emitted alongside the hub publish so callers can connect via Qt /// signals if they prefer that to subscribing on the hub. void chain_published(const fincept::services::options::OptionChain& chain); void chain_failed(const QString& topic, const QString& error); private: OptionChainService(); OptionChainService(const OptionChainService&) = delete; OptionChainService& operator=(const OptionChainService&) = delete; /// Decompose a topic of form `option:chain:::` /// into its three keys. Returns false when the topic doesn't parse. static bool parse_chain_topic(const QString& topic, QString& broker, QString& underlying, QString& expiry); /// Drive a single chain refresh — runs broker->get_quotes() on a worker /// thread, assembles rows, computes derived stats (PCR, max pain, ATM), /// then publishes the result on the hub from the UI thread. void refresh_chain(const QString& broker_id, const QString& underlying, const QString& expiry); /// Databento-specific chain refresh via databento_fno_chain.py. void refresh_chain_databento(const QString& underlying, const QString& expiry); /// Fyers-specific chain refresh via /data/options-chain-v3 endpoint. /// Returns OI, Greeks, IV, volume in a single REST call — no Python needed. void refresh_chain_fyers(const QString& broker_id, const QString& underlying, const QString& expiry); /// Hardcoded list of popular US equity option underlyings. static QStringList databento_underlyings(); /// Compute max pain strike from a fully-populated chain. static double compute_max_pain(const QVector& rows); /// Compute PCR = sum(PE OI) / sum(CE OI), or 0 when total CE OI is 0. static double compute_pcr(const QVector& rows, double& total_ce_oi, double& total_pe_oi); /// Strike with smallest |strike - spot|. static double compute_atm(const QVector& rows, double spot); /// Per-token last-known spot lookup. Populated from the hub /// (DataHub::peek("market:quote:")) on demand. Falls back to the /// broker's get_quotes for the underlying on first call per session. double resolve_spot(const QString& broker_id, const QString& underlying); /// Phase 3 — Greeks/IV via py_vollib daemon. Builds the contract list /// from `chain` (skipping per-token throttled strikes which use cached /// values), submits to OptionGreeksWorker, and on success patches the /// chain rows in place + republishes on `topic`. Also publishes /// `option:atm_iv::` and per-leg `option:tick:*`. void enrich_with_greeks(const fincept::services::options::OptionChain& chain, const QString& topic); /// Push per-leg `option:tick::` BrokerQuote snapshots /// for every CE/PE row with a non-zero token. Phase 3 fan-out is /// chain-derived; broker-WS-driven publishes will replace this in a /// later phase without breaking subscribers. void publish_per_leg_ticks(const fincept::services::options::OptionChain& chain); /// Average ATM CE/PE IV after Greeks land. Publishes /// `option:atm_iv::` (decimal IV). void publish_atm_iv(const fincept::services::options::OptionChain& chain); // ── WS-first live streaming (Phase 3 fan-out) ────────────────────────── // When the broker has a native option WebSocket (Fyers today), the market // is open, and the account is connected, the chain switches from REST // polling to a live WS feed: the ATM±window option symbols are subscribed // on the account's shared AccountDataStream, and incoming ticks patch // last_chain_ in place + fan out on `option:tick::`. REST is // suppressed (see refresh()) while the feed stays fresh, and resumes as the // fallback when the feed goes stale / market closes / account disconnects. /// Subscribe the ATM±window legs of `chain` to the live WS and route ticks. /// No-op (and tears down any existing feed) when not WS-eligible. void maybe_start_ws_stream(const fincept::services::options::OptionChain& chain, const QString& topic); /// Drop the WS subscription + tick routing for the active streamed topic. void stop_ws_stream(); /// Tick handler — merges a live broker quote into last_chain_ and republishes /// the affected leg on `option:tick:*`. Filters by account + symbol map. void on_ws_quote(const QString& account_id, const QString& symbol, const fincept::trading::BrokerQuote& quote); /// True when `topic` is the streamed topic and a tick arrived within the /// staleness window (so the REST poll should stand down). bool ws_feed_fresh(const QString& topic) const; /// Risk-free rate from settings (`fno.risk_free_rate`), default 0.067 /// (RBI 91-day T-bill ballpark). Cached after first read. double risk_free_rate(); /// Time to expiry in years, actual/365. Floors at one calendar day so /// expiry-day options don't blow up the BSM model. static double compute_t_years(const QString& expiry); fincept::services::options::OptionChain last_chain_; bool hub_registered_ = false; /// Extra symbols pinned per topic (e.g. by FnoDataBridge for algo legs). /// Unioned into the WS subscription set inside maybe_start_ws_stream. QHash pinned_contracts_; /// In-flight guard per topic to avoid duplicate refresh fan-out when the /// hub scheduler races with a manual request. QHash in_flight_; /// Last spot snapshot per (broker, underlying) — refreshed each tick. QHash spot_cache_; /// Per-token last Greeks compute timestamp (ms since epoch) — used to /// throttle py_vollib invocations to ≤1/500ms per strike per side. QHash last_greeks_compute_ms_; /// Per-token IV cache (decimal) — reused when throttle skips compute. QHash iv_cache_; /// Per-token Greeks cache — reused when throttle skips compute. QHash greeks_cache_; /// Cached risk-free rate; populated on first refresh. double risk_free_rate_ = 0.0; bool risk_free_rate_loaded_ = false; /// Cached Databento expiries per underlying (session-scoped). QHash databento_expiry_cache_; // ── WS-first streaming state ─────────────────────────────────────────── QString ws_topic_; ///< chain topic currently WS-streamed ("" = none) QString ws_account_id_; ///< account whose stream we subscribed QHash ws_sym_token_; ///< bare WS option symbol → leg token qint64 ws_last_tick_ms_ = 0; ///< last WS tick arrival (epoch ms) QMetaObject::Connection ws_quote_conn_; ///< AccountDataStream::quote_updated binding QMetaObject::Connection ws_idle_conn_; ///< DataHub::topic_idle teardown binding }; } // namespace fincept::services::options