"""JQData login and canonical daily snapshot adapter. The password is accepted only from a caller or an interactive prompt. It is never written to the snapshot, logs, Obsidian, or Git. """ from __future__ import annotations import getpass import importlib import math import os from bisect import bisect_right from datetime import date, datetime, time, timezone from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from quant60.data_snapshot import ( CanonicalBarRecord, IndexMembershipRecord, build_snapshot_payload, write_data_snapshot, ) from quant60.portable_core import normalize_symbol class JQDataUnavailableError(RuntimeError): pass def load_jqdata() -> Any: try: return importlib.import_module("jqdatasdk") except ImportError as exc: raise JQDataUnavailableError( "jqdatasdk is unavailable; install requirements/jqdata.txt" ) from exc def authenticate( sdk: Any, *, username: str | None = None, password: str | None = None, interactive: bool = True, ) -> None: account = username or os.environ.get("JQDATA_USERNAME") secret = password or os.environ.get("JQDATA_PASSWORD") if interactive and not account: account = input("JQData account: ").strip() if interactive and not secret: secret = getpass.getpass("JQData password: ") if not account or not secret: raise JQDataUnavailableError( "JQData credentials are required through environment or prompt" ) auth = getattr(sdk, "auth", None) if not callable(auth): raise JQDataUnavailableError("jqdatasdk.auth is unavailable") auth(account, secret) is_auth = getattr(sdk, "is_auth", None) if callable(is_auth) and is_auth() is not True: raise JQDataUnavailableError("JQData authentication was not accepted") def _row_value(row: Any, name: str, default: Any = None) -> Any: if isinstance(row, dict): return row.get(name, default) try: return row[name] except (KeyError, TypeError): return getattr(row, name, default) def _required_row_value(row: Any, name: str) -> Any: marker = object() value = _row_value(row, name, marker) if value is marker or value is None: raise JQDataUnavailableError( f"JQData get_bars omitted required field {name}" ) return value def _as_date(value: Any) -> date: if isinstance(value, datetime): return value.date() if isinstance(value, date): return value parser = getattr(value, "date", None) if callable(parser): parsed = parser() if isinstance(parsed, date): return parsed return date.fromisoformat(str(value)[:10]) def _optional_positive_float(value: Any) -> float | None: if value is None: return None numeric = float(value) if not math.isfinite(numeric) or numeric <= 0: return None return numeric def _required_positive_float( row: Any, name: str, *, symbol: str, timestamp: Any, ) -> float: numeric = _optional_positive_float(_required_row_value(row, name)) if numeric is None: raise JQDataUnavailableError( f"invalid {name} field for {symbol} at {timestamp}" ) return numeric def _required_bool( value: Any, name: str, *, symbol: str, timestamp: Any, ) -> bool: if isinstance(value, str): raise JQDataUnavailableError( f"invalid {name} field for {symbol} at {timestamp}" ) try: numeric = float(value) except (TypeError, ValueError, OverflowError) as exc: raise JQDataUnavailableError( f"invalid {name} field for {symbol} at {timestamp}" ) from exc if not math.isfinite(numeric) or numeric not in (0.0, 1.0): raise JQDataUnavailableError( f"invalid {name} field for {symbol} at {timestamp}" ) return bool(numeric) def fetch_index_daily_snapshot( sdk: Any, *, index_symbol: str, start_date: date, end_date: date, output_dir: str | Path, retrieved_at: datetime | None = None, ) -> dict[str, str]: """Download raw, unadjusted daily bars plus PIT index membership.""" if start_date > end_date: raise ValueError("start_date must not follow end_date") retrieval_time = retrieved_at or datetime.now(timezone.utc) if ( retrieval_time.tzinfo is None or retrieval_time.utcoffset() is None ): raise ValueError("retrieved_at must include a timezone") # JQData documents a 24:00 reconciliation for daily bars. Requiring the # requested end date to precede the Shanghai retrieval date prevents an # intraday/preliminary daily bar from being labelled T_CLOSE_COMPLETE. shanghai_retrieval_date = retrieval_time.astimezone( ZoneInfo("Asia/Shanghai") ).date() if end_date >= shanghai_retrieval_date: raise JQDataUnavailableError( "end_date must precede the Asia/Shanghai retrieval date so the " "daily bar has passed JQData's 24:00 finalization boundary" ) canonical_index = normalize_symbol(index_symbol) trade_days = [ _as_date(item) for item in sdk.get_trade_days( start_date=start_date, end_date=end_date, ) ] if not trade_days: raise JQDataUnavailableError("JQData returned no trading sessions") if trade_days != sorted(set(trade_days)): raise JQDataUnavailableError( "JQData returned a non-unique or unordered trading calendar" ) get_all_trade_days = getattr(sdk, "get_all_trade_days", None) if not callable(get_all_trade_days): raise JQDataUnavailableError( "jqdatasdk.get_all_trade_days is required to resolve the " "next trading session" ) full_calendar = [_as_date(item) for item in get_all_trade_days()] if full_calendar != sorted(set(full_calendar)): raise JQDataUnavailableError( "JQData returned a non-unique or unordered full trading calendar" ) expected_trade_days = [ item for item in full_calendar if start_date <= item <= end_date ] if trade_days != expected_trade_days: raise JQDataUnavailableError( "JQData trading-calendar APIs returned inconsistent sessions" ) next_index = bisect_right(full_calendar, end_date) if next_index >= len(full_calendar): raise JQDataUnavailableError( "JQData did not return the unique next trading session " "after end_date" ) next_trading_session = full_calendar[next_index] memberships: list[IndexMembershipRecord] = [] symbols: set[str] = set() for trading_date in trade_days: members = sdk.get_index_stocks( canonical_index, date=trading_date, ) if not members: raise JQDataUnavailableError( f"no index members on {trading_date.isoformat()}" ) for member in sorted(members): canonical = normalize_symbol(member) symbols.add(canonical) memberships.append( IndexMembershipRecord( effective_date=trading_date, index_symbol=canonical_index, member_symbol=canonical, source="jqdata", ) ) get_extras = getattr(sdk, "get_extras", None) if not callable(get_extras): raise JQDataUnavailableError( "jqdatasdk.get_extras is required for PIT is_st status" ) status_frame = get_extras( "is_st", sorted(symbols), start_date=start_date, end_date=end_date, df=True, ) if status_frame is None or not hasattr(status_frame, "iterrows"): raise JQDataUnavailableError("invalid get_extras is_st result") pit_is_st: dict[tuple[date, str], bool] = {} for timestamp, row in status_frame.iterrows(): trading_date = _as_date(timestamp) for symbol in sorted(symbols): value = _required_row_value(row, symbol) pit_is_st[(trading_date, symbol)] = _required_bool( value, "is_st", symbol=symbol, timestamp=timestamp, ) bars: list[CanonicalBarRecord] = [] fields = [ "date", "open", "high", "low", "close", "volume", "money", "factor", "paused", "high_limit", "low_limit", "pre_close", ] get_bars = getattr(sdk, "get_bars", None) if not callable(get_bars): raise JQDataUnavailableError( "jqdatasdk.get_bars is required for raw price, factor, " "price-limit and previous-close fields" ) for symbol in sorted(symbols): frame = get_bars( symbol, start_dt=start_date, # A date-only end_dt is interpreted as midnight by data APIs and # can exclude the requested final trading session. Make the # inclusive daily boundary explicit. end_dt=datetime.combine(end_date, time(23, 59, 59)), unit="1d", fields=fields, skip_paused=False, include_now=True, fq_ref_date=None, df=True, ) if frame is None or not hasattr(frame, "iterrows"): raise JQDataUnavailableError( f"invalid get_bars result for {symbol}" ) for timestamp, row in frame.iterrows(): trading_date = _as_date(_required_row_value(row, "date")) status_key = (trading_date, symbol) if status_key not in pit_is_st: raise JQDataUnavailableError( "JQData get_extras omitted PIT is_st status for " f"{symbol} at {trading_date.isoformat()}" ) raw_paused = _required_row_value(row, "paused") try: paused_numeric = float(raw_paused) except (TypeError, ValueError) as exc: raise JQDataUnavailableError( f"invalid paused field for {symbol} at {timestamp}" ) from exc if ( not math.isfinite(paused_numeric) or paused_numeric not in (0.0, 1.0) ): raise JQDataUnavailableError( f"invalid paused field for {symbol} at {timestamp}" ) paused = bool(paused_numeric) volume = int(_required_row_value(row, "volume")) money = float(_required_row_value(row, "money")) if not math.isfinite(money) or money < 0: raise JQDataUnavailableError( f"invalid money field for {symbol} at {timestamp}" ) factor = _required_positive_float( row, "factor", symbol=symbol, timestamp=timestamp, ) bars.append( CanonicalBarRecord( trading_date=trading_date, symbol=symbol, open=float(_required_row_value(row, "open")), high=float(_required_row_value(row, "high")), low=float(_required_row_value(row, "low")), close=float(_required_row_value(row, "close")), volume=volume, money=money, paused=paused, source="jqdata", is_st=pit_is_st[status_key], adjustment="none", adjustment_factor=factor, previous_close=_required_positive_float( row, "pre_close", symbol=symbol, timestamp=timestamp, ), limit_up=_required_positive_float( row, "high_limit", symbol=symbol, timestamp=timestamp, ), limit_down=_required_positive_float( row, "low_limit", symbol=symbol, timestamp=timestamp, ), ) ) version = str(getattr(sdk, "__version__", "unknown")) payload = build_snapshot_payload( bars=bars, memberships=memberships, provider="jqdata", provider_version=version, retrieved_at=retrieval_time, next_trading_session=next_trading_session, query={ "index_symbol": canonical_index, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "trading_sessions": [ *(item.isoformat() for item in trade_days), next_trading_session.isoformat(), ], "calendar_api": "get_trade_days+get_all_trade_days", "next_session_policy": ( "first full-calendar session strictly after end_date" ), "api": "get_bars", "frequency": "1d", "adjustment": "none", "factor_semantics": ( "raw OHLC plus provider back-adjustment factor; " "front adjustment is derived per decision as_of" ), "skip_paused": False, "include_now": True, "pit_status_api": "get_extras:is_st", "finalization_policy": ( "end_date_before_retrieval_date_Asia_Shanghai" ), "fields": fields, }, ) return write_data_snapshot(payload, output_dir)