Dhan WebSocket Market Data System

Real-time Dhan market data capture with:

• 4-connection WebSocket streaming for spot, futures, CE options, and PE options
• Parquet tick persistence under data/ticks/
• A live market dashboard at /
• A reference-backed moneyflow dashboard at /moneyflow
• DuckDB historical analytics from market_data.duckdb overlaid with this app's live/parquet data

Current Architecture

Live feed and storage

• main.py loads config, instruments, storage, dashboard, and feed manager.
• src/feed_manager.py owns the websocket connections and dynamic option subscriptions.
• src/storage.py flushes quotes, OI, and reference ticks to parquet.
• Tick files are written under data/ticks/{YYYY-MM-DD}/.

Dashboards

• / serves the lightweight live market dashboard from src/dashboard.py.
• /moneyflow serves the full reference-style moneyflow UI from moneyflow.html through src/moneyflow_page.py.

Moneyflow pipeline

• Historical analytics come from moneyflow.py against market_data.duckdb.
• Local path/config shims live in db_paths.py and data_service.py.
• src/reference_moneyflow.py is the integration layer:
  • uses DuckDB/reference analytics for historical and baseline calculations
  • overlays current live in-memory ticks or parquet replay data for active-session freshness
  • returns the field shapes expected by the full moneyflow frontend
• src/canonical_moneyflow.py is still used as the live/parquet mirror layer.

Key Files

• config.yaml: runtime configuration
• market_data.duckdb: historical options/ohlcv analytics database
• sector_mapping.json: symbol-to-sector mapping
• moneyflow.py: reference moneyflow analytics engine
• moneyflow.html: full moneyflow frontend
• src/reference_moneyflow.py: live integration wrapper
• tests/: unit and integration-style tests

Requirements

• Python 3.10+
• Dhan client id and access token
• Local DuckDB file at market_data.duckdb

Install dependencies:

pip install -r requirements.txt

Configuration

Runtime configuration is read from config.yaml, not from .env.

Important keys:

dhan:
  client_id: "..."
  access_token: "..."

storage:
  data_dir: "./data"
  flush_interval_seconds: 15
  retention_days: 7

feed:
  options_strike_range: 10
  options_max_expiries: 1
  atm_refresh_seconds: 60

dashboard:
  enabled: true
  host: "127.0.0.1"
  port: 8765
  batch_interval_ms: 500
  moneyflow_refresh_ms: 1500
  sector_map_path: ""

Optional moneyflow-specific environment overrides:

$env:WS_MONEYFLOW_DB_PATH = "C:\path\to\market_data.duckdb"
$env:WS_MONEYFLOW_INTRADAY_DB_PATH = "C:\path\to\live_market.duckdb"
$env:WS_MONEYFLOW_ATM_WINDOW = "10"

Default behavior:

• WS_MONEYFLOW_DB_PATH defaults to ./market_data.duckdb
• WS_MONEYFLOW_INTRADAY_DB_PATH defaults to ./data/live_market.duckdb
• If no intraday DuckDB exists, the reference engine falls back to the base historical DB

Running

Start the full system:

python main.py

Then open:

• Live dashboard: http://127.0.0.1:8765/
• Moneyflow dashboard: http://127.0.0.1:8765/moneyflow

Stop with Ctrl+C.

HTTP Endpoints

Live dashboard

• GET /
• GET /api/snapshot
• GET /events
• GET /api/stats

Moneyflow dashboard

• GET /moneyflow
• GET /api/moneyflow/snapshot
• GET /api/moneyflow/dates
• GET /api/moneyflow/expiries
• GET /api/moneyflow/all
• GET /api/moneyflow/summary
• GET /api/moneyflow/symbols
• GET /api/moneyflow/sectors
• GET /api/moneyflow/unusual
• GET /api/moneyflow/history
• GET /api/moneyflow/live-compute
• GET /api/moneyflow/parquet-compute
• GET /api/moneyflow/parquet-stream
• GET /api/moneyflow/prev-zscores
• GET /api/ws-live-ohlcv
• GET /api/ws-live-ohlcv-snapshot

Data Behavior

Historical moneyflow

• Uses DuckDB/reference analytics directly from market_data.duckdb.
• Suitable for previous trading dates and history charts.

Active-session moneyflow

• Prefers the live in-memory engine when websocket ticks are available.
• Falls back to parquet replay when live in-memory data is unavailable.
• Preserves reference-derived fields such as MFS, z-scores, sector aggregates, and history-compatible symbol rows.

OHLCV/OI stream

• For the latest available trade date, /api/ws-live-ohlcv serves current live OHLCV plus aggregated CE/PE OI.
• For historical dates, it falls back to the parquet snapshot path.

Testing

Run the full test suite:

python -m unittest discover -s tests -p "test_*.py"

Notable coverage:

• tests/test_dashboard.py: dashboard behavior and SSE/disconnect handling
• tests/test_canonical_moneyflow.py: parquet/live canonical parity checks
• tests/test_reference_moneyflow.py: reference frontend contract, live overlay merge, and OHLCV CE/PE OI snapshot behavior

Notes

• The first cold DuckDB moneyflow query can be noticeably slower than subsequent cached queries.
• The first cold parquet replay for a date can also be expensive; repeated calls are much faster because the replay layer caches results.
• Sector mapping is loaded from sector_mapping.json in the repo root when present.