You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
802 lines
35 KiB
Python
802 lines
35 KiB
Python
"""Detect new events from the pipeline DB and enrich them for paper trading.
|
|
|
|
Mirrors the enrichment pattern in SnapshotStore._async_load() but targets
|
|
a single execution_date rather than loading an entire parquet file.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import datetime as dt
|
|
from typing import Any
|
|
|
|
from libs.backtest.domain import BacktestConfig
|
|
from libs.common.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class EventDetector:
|
|
"""Queries the pipeline DB for events executing on a given date and enriches them."""
|
|
|
|
def __init__(
|
|
self,
|
|
db_dsn: str,
|
|
oracle_url: str,
|
|
bars_cache: dict[str, dict[dt.date, dict]] | None = None,
|
|
) -> None:
|
|
self._db_dsn = db_dsn
|
|
self._oracle_url = oracle_url
|
|
self._bars_cache = bars_cache # pre-fetched bars from backtest_sim
|
|
self._db_unavailable: bool = False # circuit breaker: skip after first failure
|
|
self._screener_unavailable: bool = False # circuit breaker for screener API
|
|
self._company_cache: dict[str, dict[str, Any]] = {} # symbol -> {sector, market_cap}
|
|
self._screener_cache: dict[str, float | None] | None = None # cached screener results
|
|
|
|
@staticmethod
|
|
def _compute_score(row: dict[str, Any], config: BacktestConfig) -> float:
|
|
"""Compute score using the config's scoring_model — matches backtester."""
|
|
import libs.backtest.scoring as _scoring
|
|
model = config.signal.scoring_model
|
|
|
|
# Derive function name from model name to stay in sync with new models
|
|
# without requiring manual updates here.
|
|
# "return_max_long_v13e" → compute_return_max_long_score_v13e
|
|
# "return_max_long_v1" → compute_return_max_long_score (legacy, no suffix)
|
|
# "pead" / "patient_drift" / "microstructure" → compute_{model}_score
|
|
fn: Any = None
|
|
if model.startswith("return_max_long_"):
|
|
suffix = model[len("return_max_long_"):]
|
|
fn_name = "compute_return_max_long_score" if suffix == "v1" else f"compute_return_max_long_score_{suffix}"
|
|
fn = getattr(_scoring, fn_name, None)
|
|
if fn is None:
|
|
fn = getattr(_scoring, f"compute_{model}_score", None)
|
|
if fn is None:
|
|
fn = _scoring.compute_entry_score
|
|
return fn(row)
|
|
|
|
async def get_candidates_for_date(
|
|
self,
|
|
execution_date: dt.date,
|
|
config: BacktestConfig,
|
|
convention: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Return enriched candidate rows for execution_date.
|
|
|
|
convention: 'reaction_close' for same-day close entries,
|
|
'next_open_after_reaction_close' for next-open entries.
|
|
None fetches both (legacy / debugging only).
|
|
|
|
Steps:
|
|
1. Query Event table for events with entry_date == execution_date
|
|
2. Fetch SymbolMaster tickers
|
|
3. Enrich with Alpaca price data (bars, avg_dollar_volume, ATR-14)
|
|
4. Enrich with Oracle sector data
|
|
5. Compute entry_price_est and score
|
|
"""
|
|
raw_rows = await self._fetch_events_for_date(execution_date, convention=convention)
|
|
if not raw_rows:
|
|
logger.debug("event_detector_no_events", date=execution_date.isoformat())
|
|
return []
|
|
enriched = await self._enrich_raw_rows(raw_rows, bar_end_date=execution_date, config=config)
|
|
logger.info(
|
|
"event_detector_candidates_ready",
|
|
date=execution_date.isoformat(),
|
|
count=len(enriched),
|
|
)
|
|
return enriched
|
|
|
|
async def get_candidates_for_lookback(
|
|
self,
|
|
today: dt.date,
|
|
start_date: dt.date,
|
|
config: BacktestConfig,
|
|
) -> list[dict[str, Any]]:
|
|
"""Return enriched candidate rows for events with entry_date in [start_date, today).
|
|
|
|
Used on daemon startup to pick up events from previous trading days that are
|
|
still within their max_holding_days window. Each returned row has:
|
|
- is_lookback_entry=True
|
|
- lookback_days_elapsed=N (trading days since the original execution_date)
|
|
"""
|
|
raw_rows = await self._fetch_events_for_date_range(start_date, today)
|
|
if not raw_rows:
|
|
logger.debug(
|
|
"event_detector_no_lookback_events",
|
|
start=start_date.isoformat(),
|
|
end=today.isoformat(),
|
|
)
|
|
return []
|
|
enriched = await self._enrich_raw_rows(raw_rows, bar_end_date=today, config=config)
|
|
# Annotate with lookback metadata
|
|
from libs.backtest.calendar import get_trading_days
|
|
_elapsed_cache: dict[dt.date, int] = {}
|
|
for row in enriched:
|
|
raw_exec = row.get("execution_date") or row.get("entry_date")
|
|
if raw_exec:
|
|
exec_date = raw_exec if isinstance(raw_exec, dt.date) else dt.date.fromisoformat(str(raw_exec))
|
|
if exec_date not in _elapsed_cache:
|
|
tdays = get_trading_days(exec_date, today)
|
|
_elapsed_cache[exec_date] = max(0, len(tdays) - 1)
|
|
row["is_lookback_entry"] = True
|
|
row["lookback_days_elapsed"] = _elapsed_cache[exec_date]
|
|
logger.info(
|
|
"event_detector_lookback_candidates_ready",
|
|
start=start_date.isoformat(),
|
|
today=today.isoformat(),
|
|
count=len(enriched),
|
|
)
|
|
return enriched
|
|
|
|
async def _enrich_raw_rows(
|
|
self,
|
|
raw_rows: list[dict[str, Any]],
|
|
bar_end_date: dt.date,
|
|
config: BacktestConfig,
|
|
) -> list[dict[str, Any]]:
|
|
"""Enrich raw DB rows with bars, market features, and scores.
|
|
|
|
Shared by get_candidates_for_date and get_candidates_for_lookback.
|
|
bar_end_date is the upper bound for bar fetches (usually execution_date or today).
|
|
"""
|
|
unique_symbols = sorted({
|
|
str(r.get("symbol", "")).upper()
|
|
for r in raw_rows
|
|
if r.get("symbol")
|
|
})
|
|
|
|
# Fetch 120 days of bars — enough for 60d features (entropy, Hurst, gravitational pull)
|
|
# which need ~65 trading days (~91 calendar days) before the reaction date.
|
|
bar_start = bar_end_date - dt.timedelta(days=120)
|
|
bars_by_symbol, avg_dvol, atr_by_symbol = await self._fetch_enrichment_data(
|
|
unique_symbols, bar_start, bar_end_date
|
|
)
|
|
company_info = await self._fetch_company_info(unique_symbols)
|
|
screener_mcaps = await self._fetch_screener_market_caps(unique_symbols)
|
|
|
|
enriched_rows: list[dict[str, Any]] = []
|
|
for row in raw_rows:
|
|
sym = str(row.get("symbol", "")).upper()
|
|
if not sym:
|
|
continue
|
|
|
|
info = company_info.get(sym, {})
|
|
enriched = dict(row)
|
|
enriched["symbol"] = sym
|
|
# Use Oracle-computed avg_dvol; fall back to stored avg_dollar_volume_20d
|
|
# from feature_json when Oracle enrichment returned nothing.
|
|
oracle_adv = avg_dvol.get(sym)
|
|
enriched["avg_dollar_volume"] = (
|
|
oracle_adv if oracle_adv
|
|
else float(row.get("avg_dollar_volume_20d") or 0.0)
|
|
)
|
|
# Prefer atr_14 stored in DB feature_json (computed by feature builder with
|
|
# event_date+5d bars, giving stable post-event ATR). Fall back to Oracle-computed
|
|
# value only if not already present — avoids inflating ATR with the large
|
|
# earnings reaction-day range when bars end exactly at execution_date.
|
|
if not enriched.get("atr_14"):
|
|
enriched["atr_14"] = atr_by_symbol.get(sym)
|
|
# Enrich reaction_day_low / reaction_day_high from Oracle bars.
|
|
# These are NOT stored in DB feature_json but ARE used by compute_stop_price()
|
|
# to tighten stops when reaction_day_low > atr_stop (matches snapshot pipeline).
|
|
if not enriched.get("reaction_day_low") or not enriched.get("reaction_day_high"):
|
|
sym_bars = bars_by_symbol.get(sym, {})
|
|
reaction_date_raw = enriched.get("reaction_date")
|
|
if reaction_date_raw:
|
|
rd = _parse_date(reaction_date_raw)
|
|
if rd and rd in sym_bars:
|
|
if not enriched.get("reaction_day_low"):
|
|
enriched["reaction_day_low"] = sym_bars[rd].get("low")
|
|
if not enriched.get("reaction_day_high"):
|
|
enriched["reaction_day_high"] = sym_bars[rd].get("high")
|
|
|
|
# Compute market features from Oracle bars if missing in DB feature_json.
|
|
# These can be None when the feature builder ran before reaction-day bars settled.
|
|
sym_bars = bars_by_symbol.get(sym, {})
|
|
rd = _parse_date(enriched.get("reaction_date"))
|
|
if rd and rd in sym_bars:
|
|
sorted_dates = sorted(sym_bars.keys())
|
|
try:
|
|
rd_idx = sorted_dates.index(rd)
|
|
except ValueError:
|
|
rd_idx = -1
|
|
if rd_idx > 0:
|
|
prior_dates = sorted_dates[max(0, rd_idx - 20):rd_idx]
|
|
prev_bar = sym_bars[sorted_dates[rd_idx - 1]]
|
|
reaction_bar = sym_bars[rd]
|
|
|
|
if not enriched.get("volume_ratio_20d") and prior_dates:
|
|
avg_vol = sum(sym_bars[d]["volume"] for d in prior_dates) / len(prior_dates)
|
|
if avg_vol > 0:
|
|
vr = reaction_bar["volume"] / avg_vol
|
|
enriched["volume_ratio_20d"] = vr
|
|
|
|
if not enriched.get("reaction_day_return") and prev_bar["close"]:
|
|
enriched["reaction_day_return"] = (
|
|
(reaction_bar["close"] - prev_bar["close"]) / prev_bar["close"]
|
|
)
|
|
|
|
if not enriched.get("close_location"):
|
|
rng = reaction_bar["high"] - reaction_bar["low"]
|
|
if rng > 0:
|
|
enriched["close_location"] = (
|
|
(reaction_bar["close"] - reaction_bar["low"]) / rng
|
|
)
|
|
|
|
if not enriched.get("gap_size") and prev_bar["close"]:
|
|
enriched["gap_size"] = (
|
|
(reaction_bar["open"] - prev_bar["close"]) / prev_bar["close"]
|
|
)
|
|
|
|
enriched["sector"] = info.get("sector", "UNKNOWN")
|
|
if "market_cap_proxy" not in enriched or enriched["market_cap_proxy"] is None:
|
|
# Prefer screener market_cap (same source as snapshot export pipeline)
|
|
market_cap = screener_mcaps.get(sym)
|
|
if market_cap is None:
|
|
market_cap = info.get("market_cap")
|
|
enriched["market_cap_proxy"] = market_cap
|
|
|
|
# entry_price_est: use reaction-day close price
|
|
if "entry_price_est" not in enriched or not enriched["entry_price_est"]:
|
|
event_close = enriched.get("event_close")
|
|
if event_close:
|
|
enriched["entry_price_est"] = float(event_close)
|
|
else:
|
|
# Fall back to latest available close from bars
|
|
sym_bars = bars_by_symbol.get(sym, {})
|
|
reaction_date_raw = enriched.get("reaction_date")
|
|
if reaction_date_raw:
|
|
rd = _parse_date(reaction_date_raw)
|
|
if rd and rd in sym_bars:
|
|
enriched["entry_price_est"] = sym_bars[rd].get("close", 0.0)
|
|
|
|
if not enriched.get("entry_price_est"):
|
|
# Final fallback: use the most recent available close
|
|
# (needed for same-day pending events where reaction_date bar not yet available)
|
|
sym_bars = bars_by_symbol.get(sym, {})
|
|
if sym_bars:
|
|
latest_d = max(sym_bars.keys())
|
|
enriched["entry_price_est"] = sym_bars[latest_d].get("close", 0.0)
|
|
|
|
if not enriched.get("entry_price_est"):
|
|
logger.debug(
|
|
"event_detector_skip_no_price",
|
|
symbol=sym,
|
|
event_id=enriched.get("event_id"),
|
|
)
|
|
continue
|
|
|
|
# Compute tier2/tier3 features from bars if missing from DB.
|
|
# The DB FeatureSnapshot only stores basic NLP/event features; technical
|
|
# features (entropy, Hurst, gravitational pull, etc.) are computed only in
|
|
# the Parquet enrichment pipeline. Reproduce them here from Oracle bars.
|
|
# Use event_date (pre-event baseline) to match the Parquet pipeline exactly.
|
|
# Fall back to reaction_date if event_date is unavailable or not a trading day.
|
|
sym_bars_for_features = bars_by_symbol.get(sym, {})
|
|
_event_date_raw = enriched.get("event_date")
|
|
_ed_candidate = _parse_date(_event_date_raw)
|
|
# event_date must be present in bars (i.e., a trading day) to use it
|
|
if _ed_candidate and _ed_candidate in sym_bars_for_features:
|
|
rd_for_features = _ed_candidate
|
|
else:
|
|
rd_for_features = _parse_date(enriched.get("reaction_date"))
|
|
if rd_for_features and sym_bars_for_features:
|
|
from libs.features.market_features import (
|
|
avg_dollar_volume_20d as _adv20d_fn,
|
|
pre_event_bb_position as _bb_pos_fn,
|
|
pre_event_entropy as _entropy_fn,
|
|
pre_event_gravitational_pull as _grav_pull_fn,
|
|
pre_event_hurst as _hurst_fn,
|
|
pre_event_market_temperature as _mkt_temp_fn,
|
|
)
|
|
from libs.oracle_client.models import PriceBar as _OraclePriceBar
|
|
|
|
_price_bars = [
|
|
_OraclePriceBar(
|
|
date=d.isoformat(),
|
|
open=float(b.get("open", 0)),
|
|
high=float(b.get("high", 0)),
|
|
low=float(b.get("low", 0)),
|
|
close=float(b.get("close", 0)),
|
|
volume=int(b.get("volume", 0)),
|
|
)
|
|
for d, b in sorted(sym_bars_for_features.items())
|
|
]
|
|
_rd_str = rd_for_features.isoformat()
|
|
|
|
if enriched.get("avg_dollar_volume_20d") is None:
|
|
enriched["avg_dollar_volume_20d"] = _adv20d_fn(_price_bars, _rd_str)
|
|
if enriched.get("pre_event_bb_position") is None:
|
|
enriched["pre_event_bb_position"] = _bb_pos_fn(_price_bars, _rd_str)
|
|
if enriched.get("pre_event_hurst_60d") is None:
|
|
enriched["pre_event_hurst_60d"] = _hurst_fn(_price_bars, _rd_str)
|
|
if enriched.get("pre_event_entropy_60d") is None:
|
|
enriched["pre_event_entropy_60d"] = _entropy_fn(_price_bars, _rd_str)
|
|
if enriched.get("pre_event_gravitational_pull") is None:
|
|
enriched["pre_event_gravitational_pull"] = _grav_pull_fn(_price_bars, _rd_str)
|
|
if enriched.get("pre_event_market_temperature") is None:
|
|
enriched["pre_event_market_temperature"] = _mkt_temp_fn(_price_bars, _rd_str)
|
|
|
|
# Compute score using config's scoring model for consistency with
|
|
# BacktestRunner. Only use config model when event_v1 features are
|
|
# present (parse_confidence_overall etc.), otherwise the model's hard
|
|
# gates reject events with incomplete features.
|
|
if enriched.get("parse_confidence_overall") is not None:
|
|
enriched["score"] = self._compute_score(enriched, config)
|
|
elif "score" not in enriched or enriched.get("score") is None:
|
|
from libs.backtest.scoring import compute_entry_score
|
|
enriched["score"] = compute_entry_score(enriched)
|
|
|
|
enriched_rows.append(enriched)
|
|
|
|
return enriched_rows
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# DB queries
|
|
# ------------------------------------------------------------------ #
|
|
|
|
async def _fetch_events_for_date(
|
|
self,
|
|
execution_date: dt.date,
|
|
convention: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Query EventLabel + FeatureSnapshot + Event + SymbolMaster for execution_date.
|
|
|
|
entry_date lives in EventLabel, not Event. Features live in FeatureSnapshot.feature_json.
|
|
convention: filter by entry_convention ('reaction_close' or 'next_open_after_reaction_close').
|
|
"""
|
|
if self._db_unavailable:
|
|
return []
|
|
try:
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from libs.db.models import Event, EventLabel, FeatureSnapshot, SymbolMaster
|
|
|
|
engine = create_async_engine(
|
|
self._db_dsn, echo=False,
|
|
connect_args={"timeout": 5},
|
|
)
|
|
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
_UTC = __import__("zoneinfo").ZoneInfo("UTC")
|
|
|
|
async with async_session() as session:
|
|
# Join: Event + EventLabel (entry_date) + FeatureSnapshot (features) + SymbolMaster
|
|
# Order by snapshot created_at ASC so newer snapshots override older ones when merging
|
|
stmt = (
|
|
select(Event, EventLabel, FeatureSnapshot, SymbolMaster)
|
|
.join(EventLabel, Event.event_id == EventLabel.event_id)
|
|
.join(FeatureSnapshot, Event.event_id == FeatureSnapshot.event_id)
|
|
.outerjoin(SymbolMaster, Event.symbol_id == SymbolMaster.symbol_id)
|
|
.where(EventLabel.entry_date == execution_date)
|
|
.where(EventLabel.label_status.in_(["ok", "truncated", "pending"]))
|
|
.order_by(FeatureSnapshot.created_at_utc.asc())
|
|
)
|
|
if convention is not None:
|
|
stmt = stmt.where(EventLabel.entry_convention == convention)
|
|
rows = (await session.execute(stmt)).all()
|
|
|
|
await engine.dispose()
|
|
|
|
# Group by event_id — merge ALL FeatureSnapshot.feature_json per event.
|
|
# The feature builder creates multiple snapshots per event with different feature types
|
|
# (market features, NLP features, fundamental features). We need all of them merged.
|
|
# Ordered ASC so newer keys override older ones.
|
|
event_meta: dict[str, tuple] = {} # event_id -> (event, label, sym)
|
|
event_features: dict[str, dict] = {} # event_id -> merged feature_json
|
|
|
|
for event, label, snapshot, sym in rows:
|
|
eid = event.event_id
|
|
if sym is None or not sym.ticker:
|
|
continue
|
|
if eid not in event_meta:
|
|
event_meta[eid] = (event, label, sym)
|
|
event_features[eid] = {}
|
|
# Merge: newer snapshot keys override older ones (ASC order)
|
|
event_features[eid].update(snapshot.feature_json or {})
|
|
|
|
result: list[dict[str, Any]] = []
|
|
for eid, (event, label, sym) in event_meta.items():
|
|
ts = event.filed_at_utc
|
|
if ts is None and event.event_date is not None:
|
|
ts = dt.datetime.combine(
|
|
event.event_date, dt.time(21, 0), tzinfo=_UTC
|
|
)
|
|
|
|
row: dict[str, Any] = {
|
|
"event_id": eid,
|
|
"symbol": sym.ticker,
|
|
"issuer_id": event.issuer_id,
|
|
"event_type": event.event_type or "",
|
|
"event_direction": event.event_direction or "",
|
|
"event_date": event.event_date,
|
|
"event_timestamp": ts,
|
|
"reaction_date": label.reaction_date,
|
|
"entry_date": label.entry_date,
|
|
"entry_convention": label.entry_convention,
|
|
"execution_date": execution_date,
|
|
"label_status": label.label_status,
|
|
# Merged features from ALL FeatureSnapshots for this event
|
|
**event_features[eid],
|
|
}
|
|
result.append(row)
|
|
|
|
logger.debug(
|
|
"event_detector_db_fetched",
|
|
date=execution_date.isoformat(),
|
|
count=len(result),
|
|
)
|
|
return result
|
|
|
|
except Exception as exc:
|
|
if not self._db_unavailable:
|
|
err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__
|
|
logger.warning("event_detector_db_fetch_failed", error=err_msg)
|
|
self._db_unavailable = True
|
|
return []
|
|
|
|
async def _fetch_events_for_date_range(
|
|
self,
|
|
start_date: dt.date,
|
|
end_date: dt.date,
|
|
) -> list[dict[str, Any]]:
|
|
"""Query EventLabel rows where entry_date in [start_date, end_date).
|
|
|
|
Used for lookback entry: surfaces events that fired before the daemon
|
|
started but are still within their max_holding_days window.
|
|
"""
|
|
if self._db_unavailable:
|
|
return []
|
|
try:
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from libs.db.models import Event, EventLabel, FeatureSnapshot, SymbolMaster
|
|
|
|
engine = create_async_engine(
|
|
self._db_dsn, echo=False,
|
|
connect_args={"timeout": 5},
|
|
)
|
|
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
_UTC = __import__("zoneinfo").ZoneInfo("UTC")
|
|
|
|
async with async_session() as session:
|
|
stmt = (
|
|
select(Event, EventLabel, FeatureSnapshot, SymbolMaster)
|
|
.join(EventLabel, Event.event_id == EventLabel.event_id)
|
|
.join(FeatureSnapshot, Event.event_id == FeatureSnapshot.event_id)
|
|
.outerjoin(SymbolMaster, Event.symbol_id == SymbolMaster.symbol_id)
|
|
.where(EventLabel.entry_date >= start_date)
|
|
.where(EventLabel.entry_date < end_date)
|
|
.where(EventLabel.label_status.in_(["ok", "truncated", "pending"]))
|
|
.order_by(FeatureSnapshot.created_at_utc.asc())
|
|
)
|
|
rows = (await session.execute(stmt)).all()
|
|
|
|
await engine.dispose()
|
|
|
|
event_meta: dict[str, tuple] = {}
|
|
event_features: dict[str, dict] = {}
|
|
|
|
for event, label, snapshot, sym in rows:
|
|
eid = event.event_id
|
|
if sym is None or not sym.ticker:
|
|
continue
|
|
if eid not in event_meta:
|
|
event_meta[eid] = (event, label, sym)
|
|
event_features[eid] = {}
|
|
event_features[eid].update(snapshot.feature_json or {})
|
|
|
|
result: list[dict[str, Any]] = []
|
|
for eid, (event, label, sym) in event_meta.items():
|
|
ts = event.filed_at_utc
|
|
if ts is None and event.event_date is not None:
|
|
ts = dt.datetime.combine(
|
|
event.event_date, dt.time(21, 0), tzinfo=_UTC
|
|
)
|
|
row: dict[str, Any] = {
|
|
"event_id": eid,
|
|
"symbol": sym.ticker,
|
|
"issuer_id": event.issuer_id,
|
|
"event_type": event.event_type or "",
|
|
"event_direction": event.event_direction or "",
|
|
"event_date": event.event_date,
|
|
"event_timestamp": ts,
|
|
"reaction_date": label.reaction_date,
|
|
"entry_date": label.entry_date,
|
|
"entry_convention": label.entry_convention,
|
|
"execution_date": label.entry_date, # use entry_date as execution_date
|
|
"label_status": label.label_status,
|
|
**event_features[eid],
|
|
}
|
|
result.append(row)
|
|
|
|
logger.debug(
|
|
"event_detector_db_range_fetched",
|
|
start=start_date.isoformat(),
|
|
end=end_date.isoformat(),
|
|
count=len(result),
|
|
)
|
|
return result
|
|
|
|
except Exception as exc:
|
|
if not self._db_unavailable:
|
|
err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__
|
|
logger.warning("event_detector_db_range_fetch_failed", error=err_msg)
|
|
self._db_unavailable = True
|
|
return []
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Enrichment helpers
|
|
# ------------------------------------------------------------------ #
|
|
|
|
async def _fetch_enrichment_data(
|
|
self,
|
|
symbols: list[str],
|
|
start: dt.date,
|
|
end: dt.date,
|
|
concurrency: int = 16,
|
|
) -> tuple[
|
|
dict[str, dict[dt.date, dict[str, Any]]], # bars_by_symbol
|
|
dict[str, float], # avg_dvol
|
|
dict[str, float | None], # atr_14
|
|
]:
|
|
if not symbols:
|
|
return {}, {}, {}
|
|
|
|
# Fast path: use pre-fetched bars_cache (backtest mode)
|
|
if self._bars_cache is not None:
|
|
return self._enrichment_from_cache(symbols, start, end)
|
|
|
|
try:
|
|
from libs.oracle_client import OracleClient, PriceService
|
|
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] = {}
|
|
avg_dvol: dict[str, float] = {}
|
|
atr_14_map: dict[str, float | None] = {}
|
|
semaphore = asyncio.Semaphore(concurrency)
|
|
|
|
async with OracleClient(base_url=self._oracle_url) as client:
|
|
svc = PriceService(client)
|
|
|
|
async def _fetch_one(sym: str) -> None:
|
|
async with semaphore:
|
|
try:
|
|
resp = await svc.get_daily_bars(
|
|
sym,
|
|
start=start.isoformat(),
|
|
end=end.isoformat(),
|
|
)
|
|
date_map: dict[dt.date, dict[str, Any]] = {}
|
|
dollar_vols: list[float] = []
|
|
closes: list[float] = []
|
|
for bar in resp.bars:
|
|
d = dt.date.fromisoformat(bar.date)
|
|
b = {
|
|
"date": d,
|
|
"open": bar.open,
|
|
"high": bar.high,
|
|
"low": bar.low,
|
|
"close": bar.close,
|
|
"volume": bar.volume,
|
|
}
|
|
date_map[d] = b
|
|
dollar_vols.append(bar.close * bar.volume)
|
|
closes.append(bar.close)
|
|
|
|
bars_by_symbol[sym] = date_map
|
|
last_20 = dollar_vols[-20:]
|
|
avg_dvol[sym] = sum(last_20) / len(last_20) if last_20 else 0.0
|
|
atr_14_map[sym] = _compute_atr14_from_dicts(list(date_map.values()))
|
|
except Exception as sym_exc:
|
|
err_msg = f"{type(sym_exc).__name__}: {sym_exc}" if str(sym_exc) else type(sym_exc).__name__
|
|
# "Not found" is expected for delisted/renamed symbols — debug only
|
|
if "not found" in str(sym_exc).lower():
|
|
logger.debug("event_detector_price_fetch_skipped", symbol=sym, error=err_msg)
|
|
else:
|
|
logger.warning("event_detector_price_fetch_failed", symbol=sym, error=err_msg)
|
|
bars_by_symbol[sym] = {}
|
|
avg_dvol[sym] = 0.0
|
|
atr_14_map[sym] = None
|
|
|
|
await asyncio.gather(*(_fetch_one(sym) for sym in symbols))
|
|
|
|
return bars_by_symbol, avg_dvol, atr_14_map
|
|
|
|
except Exception as exc:
|
|
err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__
|
|
logger.warning("event_detector_oracle_failed", error=err_msg)
|
|
return {}, {}, {}
|
|
|
|
def _enrichment_from_cache(
|
|
self,
|
|
symbols: list[str],
|
|
start: dt.date,
|
|
end: dt.date,
|
|
) -> tuple[
|
|
dict[str, dict[dt.date, dict[str, Any]]],
|
|
dict[str, float],
|
|
dict[str, float | None],
|
|
]:
|
|
"""Compute enrichment data from pre-fetched bars_cache (no Oracle calls)."""
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] = {}
|
|
avg_dvol: dict[str, float] = {}
|
|
atr_14_map: dict[str, float | None] = {}
|
|
|
|
for sym in symbols:
|
|
all_bars = self._bars_cache.get(sym, {})
|
|
# Filter to requested date range
|
|
date_map = {d: b for d, b in all_bars.items() if start <= d <= end}
|
|
bars_by_symbol[sym] = date_map
|
|
|
|
if not date_map:
|
|
avg_dvol[sym] = 0.0
|
|
atr_14_map[sym] = None
|
|
continue
|
|
|
|
sorted_bars = [date_map[d] for d in sorted(date_map.keys())]
|
|
dollar_vols = [b["close"] * b["volume"] for b in sorted_bars]
|
|
last_20 = dollar_vols[-20:]
|
|
avg_dvol[sym] = sum(last_20) / len(last_20) if last_20 else 0.0
|
|
atr_14_map[sym] = _compute_atr14_from_dicts(sorted_bars)
|
|
|
|
return bars_by_symbol, avg_dvol, atr_14_map
|
|
|
|
async def _fetch_screener_market_caps(
|
|
self,
|
|
symbols: list[str],
|
|
) -> dict[str, float | None]:
|
|
"""Fetch market_cap for symbols via Oracle screener (batch).
|
|
|
|
Mirrors snapshot_export._resolve_universe_profile() which uses
|
|
ScreenerService to get market_cap_proxy for all universe stocks.
|
|
"""
|
|
if not symbols or self._screener_unavailable:
|
|
return {}
|
|
|
|
# Return from cache if available (screener data is static across days)
|
|
if self._screener_cache is not None:
|
|
symbol_set = {s.upper() for s in symbols}
|
|
return {s: self._screener_cache.get(s) for s in symbol_set}
|
|
|
|
symbol_set = {s.upper() for s in symbols}
|
|
result: dict[str, float | None] = {s: None for s in symbol_set}
|
|
|
|
try:
|
|
from libs.oracle_client import OracleClient, ScreenerService
|
|
|
|
async with OracleClient(base_url=self._oracle_url) as client:
|
|
svc = ScreenerService(client)
|
|
# market_cap_min=500M matches snapshot_export pattern: reduces result set
|
|
# from ~10k to ~3k stocks, avoiding HTTP 500 on large responses.
|
|
stocks = await svc.search_all_stocks(
|
|
market_cap_min=500_000_000,
|
|
exchange="NYSE,NASDAQ,AMEX",
|
|
exclude_types="ETF,FUND,ADR,SPAC",
|
|
)
|
|
|
|
# Cache ALL screener results for future calls
|
|
self._screener_cache = {}
|
|
for stock in stocks:
|
|
sym = (stock.symbol or "").upper()
|
|
self._screener_cache[sym] = stock.market_cap
|
|
if sym in symbol_set:
|
|
result[sym] = stock.market_cap
|
|
|
|
except Exception as exc:
|
|
err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__
|
|
logger.warning("event_detector_screener_mcap_failed", error=err_msg)
|
|
self._screener_unavailable = True
|
|
|
|
return result
|
|
|
|
async def _fetch_company_info(
|
|
self,
|
|
symbols: list[str],
|
|
concurrency: int = 16,
|
|
) -> dict[str, dict[str, Any]]:
|
|
"""Fetch sector and market_cap for each symbol from Oracle (cached)."""
|
|
if not symbols:
|
|
return {}
|
|
|
|
# Only fetch symbols not yet in cache
|
|
uncached = [s for s in symbols if s not in self._company_cache]
|
|
|
|
if uncached:
|
|
semaphore = asyncio.Semaphore(concurrency)
|
|
try:
|
|
from libs.oracle_client import CompanyService, OracleClient
|
|
|
|
async with OracleClient(base_url=self._oracle_url) as client:
|
|
company_svc = CompanyService(client)
|
|
|
|
async def _fetch_one(sym: str) -> tuple[str, dict[str, Any]]:
|
|
async with semaphore:
|
|
try:
|
|
info = await company_svc.get_company(sym)
|
|
return sym, {
|
|
"sector": info.sector or "UNKNOWN",
|
|
"market_cap": info.market_cap,
|
|
}
|
|
except Exception:
|
|
return sym, {"sector": "UNKNOWN", "market_cap": None}
|
|
|
|
fetched = await asyncio.gather(*(_fetch_one(sym) for sym in uncached))
|
|
for sym, info in fetched:
|
|
self._company_cache[sym] = info
|
|
|
|
except Exception as exc:
|
|
err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__
|
|
logger.warning("event_detector_company_fetch_failed", error=err_msg)
|
|
|
|
result: dict[str, dict[str, Any]] = {}
|
|
for sym in symbols:
|
|
result[sym] = self._company_cache.get(sym, {"sector": "UNKNOWN", "market_cap": None})
|
|
return result
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Helpers
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _compute_atr14(bars: list[Any]) -> float | None:
|
|
"""Compute 14-day ATR from a list of bar objects (API response)."""
|
|
if len(bars) < 2:
|
|
return None
|
|
|
|
true_ranges: list[float] = []
|
|
for i in range(1, len(bars)):
|
|
prev_close = float(bars[i - 1].close)
|
|
high = float(bars[i].high)
|
|
low = float(bars[i].low)
|
|
tr = max(
|
|
high - low,
|
|
abs(high - prev_close),
|
|
abs(low - prev_close),
|
|
)
|
|
true_ranges.append(tr)
|
|
|
|
if not true_ranges:
|
|
return None
|
|
|
|
last_14 = true_ranges[-14:] if len(true_ranges) >= 14 else true_ranges
|
|
return sum(last_14) / len(last_14)
|
|
|
|
|
|
def _compute_atr14_from_dicts(bars: list[dict]) -> float | None:
|
|
"""Compute 14-day ATR from a list of bar dicts (cache format)."""
|
|
if len(bars) < 2:
|
|
return None
|
|
|
|
true_ranges: list[float] = []
|
|
for i in range(1, len(bars)):
|
|
prev_close = float(bars[i - 1]["close"])
|
|
high = float(bars[i]["high"])
|
|
low = float(bars[i]["low"])
|
|
tr = max(
|
|
high - low,
|
|
abs(high - prev_close),
|
|
abs(low - prev_close),
|
|
)
|
|
true_ranges.append(tr)
|
|
|
|
if not true_ranges:
|
|
return None
|
|
|
|
last_14 = true_ranges[-14:] if len(true_ranges) >= 14 else true_ranges
|
|
return sum(last_14) / len(last_14)
|
|
|
|
|
|
def _parse_date(raw: Any) -> dt.date | None:
|
|
if isinstance(raw, dt.datetime):
|
|
return raw.date()
|
|
if isinstance(raw, dt.date):
|
|
return raw
|
|
if isinstance(raw, str):
|
|
try:
|
|
return dt.date.fromisoformat(raw[:10])
|
|
except ValueError:
|
|
return None
|
|
return None
|