import logging import os import threading from dataclasses import dataclass from datetime import datetime, time, timedelta from typing import Any, Dict, List, Optional, Tuple from zoneinfo import ZoneInfo import pandas as pd from app.v1.services.zerodha.client import ZerodhaClient logger = logging.getLogger(__name__) IST = ZoneInfo("Asia/Kolkata") # NSE cash session (IST). Used only to select an intraday window for Kite historical. MARKET_OPEN_IST = time(9, 15) MARKET_CLOSE_IST = time(15, 30) @dataclass class _CacheEntry: session_ist_date: str fetched_at_utc: datetime candles: List[Dict[str, Any]] class IntradayCandleStore: """In-memory, day-scoped intraday candles. Rules enforced: - Intraday data is held in memory only. - Intraday candles are only for current trading day. - Store is flushed when IST day changes. - Never persists to DB. """ def __init__(self) -> None: self._lock = threading.Lock() self._store: Dict[Tuple[str, str], _CacheEntry] = {} # Small knob to avoid refetching too frequently per stock/timeframe. self._min_refresh_seconds = int(os.getenv("INTRADAY_MIN_REFRESH_SECONDS", "20")) def _today_ist_key(self, now_utc: Optional[datetime] = None) -> str: now_utc = now_utc or datetime.utcnow() now_ist = now_utc.replace(tzinfo=ZoneInfo("UTC")).astimezone(IST) # Choose the most recent trading session (Mon-Fri). On weekends, roll back. session_date = now_ist.date() if now_ist.time() < MARKET_OPEN_IST: session_date = session_date - timedelta(days=1) while session_date.weekday() >= 5: # Sat=5, Sun=6 session_date = session_date - timedelta(days=1) return session_date.isoformat() def _session_bounds_ist(self, now_utc: datetime) -> Tuple[datetime, datetime, str]: """Return (start_ist, end_ist, session_date_key) for intraday candles. If we are outside market hours or on a non-trading day, this returns the last trading session window. """ now_ist = now_utc.replace(tzinfo=ZoneInfo("UTC")).astimezone(IST) session_key = self._today_ist_key(now_utc) session_date = datetime.fromisoformat(session_key).date() start_ist = datetime.combine(session_date, MARKET_OPEN_IST, tzinfo=IST) end_ist = datetime.combine(session_date, MARKET_CLOSE_IST, tzinfo=IST) # If we're inside today's session, cap to "now" for fresher candles. if now_ist.date() == session_date and MARKET_OPEN_IST <= now_ist.time() <= MARKET_CLOSE_IST: end_ist = now_ist return start_ist, end_ist, session_key def flush_if_new_day(self, now_utc: Optional[datetime] = None) -> None: now_utc = now_utc or datetime.utcnow() today = self._today_ist_key(now_utc) with self._lock: # Fast path: if any entry has a different day, flush all. for _, ent in self._store.items(): if ent.session_ist_date != today: self._store.clear() return def get_intraday_candles( self, *, zerodha_client: ZerodhaClient, instrument_token: int, timeframe: str, max_candles: int = 100, now_utc: Optional[datetime] = None, ) -> List[Dict[str, Any]]: now_utc = now_utc or datetime.utcnow() self.flush_if_new_day(now_utc) tf = str(timeframe or "").strip().lower() if tf not in {"5minute", "15minute", "30minute"}: raise ValueError(f"Unsupported intraday timeframe: {timeframe}") key = (str(instrument_token), tf) session_key = self._today_ist_key(now_utc) with self._lock: ent = self._store.get(key) if ent and ent.session_ist_date == session_key: age = (now_utc - ent.fetched_at_utc).total_seconds() if age >= 0 and age < self._min_refresh_seconds: return ent.candles[-max_candles:] # Fetch candles for the most recent trading session and keep last max_candles. start_ist, end_ist, session_key = self._session_bounds_ist(now_utc) # Convert to naive datetimes (Kite accepts datetime objects; timezone handling is inconsistent). from_dt = start_ist.replace(tzinfo=None) to_dt = end_ist.replace(tzinfo=None) # Use the client wrapper to inherit retry/throttle behavior. raw = zerodha_client.get_historical_data_records( instrument_token, interval=tf, from_date=from_dt, to_date=to_dt, continuous=False, oi=False, ) # If we got nothing (common on holidays), try walking back a few sessions. # Keep it small to avoid hammering Kite. if not raw: fallback_days = int(os.getenv("INTRADAY_FALLBACK_LOOKBACK_DAYS", "3")) fallback_days = max(0, min(7, fallback_days)) if fallback_days > 0: for _ in range(fallback_days): session_date = datetime.fromisoformat(session_key).date() - timedelta(days=1) while session_date.weekday() >= 5: session_date = session_date - timedelta(days=1) start_ist = datetime.combine(session_date, MARKET_OPEN_IST, tzinfo=IST) end_ist = datetime.combine(session_date, MARKET_CLOSE_IST, tzinfo=IST) from_dt = start_ist.replace(tzinfo=None) to_dt = end_ist.replace(tzinfo=None) try: raw = zerodha_client.get_historical_data_records( instrument_token, interval=tf, from_date=from_dt, to_date=to_dt, continuous=False, oi=False, ) except Exception: raw = [] if raw: session_key = session_date.isoformat() break df = pd.DataFrame(raw or []) if not df.empty: df["date"] = pd.to_datetime(df.get("date"), errors="coerce") df = df.dropna(subset=["date"]).sort_values("date") candles = df.to_dict("records") if not df.empty else [] candles = candles[-max_candles:] if candles else [] with self._lock: self._store[key] = _CacheEntry(session_ist_date=session_key, fetched_at_utc=now_utc, candles=candles) return candles GLOBAL_INTRADAY_STORE = IntradayCandleStore()