momentum devandClaude Opus 5
11b0fcd7b7
_population_reject computed the static open gap from bars["open"].iloc[0]. Premarket that is a PREMARKET bar, so the gap is fabricated. Live on 2026-09-14: FTFT opened 3.30 against a 2.88 previous close (+14.6%, well inside the tradeable 10-35% band) but its premarket bars started near 2.52, so the filter read -12.5% and benched it at 09:12 -- before the session had even opened. It never self-corrected. The frame keeps its premarket bars all session, so every later pass recomputed the same wrong number, and _transition only emits on change, so the symbol stayed out_of_population for the day. Several of the day's rejections show negative gaps (-12.5%, -17.4%, -1.9%) that look like the same artifact. Trading was unaffected -- the trigger path does not consult watcher state, and FTFT was still evaluated and rejected on its own merits (over_extended_day, it was already +59% off the open). The damage is diagnostic: valid candidates lose their criteria trail, which is the blank-reasoning-panel problem a previous commit set out to fix. Read the open off the first is_regular_hours bar; return None (fail open) when the session has not opened, since the static gap is not yet knowable. A frame with no session flag is treated as already regular-hours-only, which is what tests and replay hand it. Also drop the pm-range=N counter from step_arm's return string. The arm step never emits a module_tick -- verified against both 2026-09-11 and 2026-09-14, where only discovery, ingestion, enrichment and triggers do -- so it reached container stdout and nothing else. Adding a tick purely to carry it would put ~4k events/day back after the population filter had just removed 88% of them. pm_range_pct is already on the armed candidate rows for the board; persist it on the approval payload if it needs to be measurable per trade. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01A2LepdJfPCfK5Q8uznjFBM
388 lines
17 KiB
Python
388 lines
17 KiB
Python
"""Intraday watcher (Milestone 4).
|
|
|
|
Pulls candidates from a watchlist provider, evaluates each with the pure
|
|
strategy engine, drives a per-symbol state machine, and emits canonical
|
|
events into the event store:
|
|
|
|
discovered -> watching -> ready | blocked (re-evaluated each tick)
|
|
|
|
Signals are debounced: a symbol emits signal_ready at most once per session
|
|
unless it falls back to blocked first.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, date
|
|
from typing import Callable, Protocol
|
|
|
|
import pandas as pd
|
|
|
|
from storage.event_schema import (
|
|
CriteriaEvaluatedEvent,
|
|
EventMode,
|
|
SignalBlockedEvent,
|
|
SignalReadyEvent,
|
|
SymbolDiscoveredEvent,
|
|
SymbolStateChangedEvent,
|
|
)
|
|
from storage.event_store import EventStore
|
|
from strategy.evaluation.setup_evaluator import evaluate_setup
|
|
from strategy.population import PopulationConfig, population_reject
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class WatchCandidate:
|
|
"""One symbol the watcher should evaluate."""
|
|
|
|
symbol: str
|
|
last_price: float | None = None
|
|
previous_close: float | None = None
|
|
avg_daily_volume: float | None = None
|
|
source: str = "research"
|
|
|
|
|
|
class WatchlistProvider(Protocol):
|
|
"""Source of candidates and their session bars."""
|
|
|
|
def get_candidates(self, session_date: date) -> list[WatchCandidate]: ...
|
|
|
|
def get_bars(self, symbol: str, session_date: date) -> pd.DataFrame: ...
|
|
|
|
|
|
@dataclass
|
|
class WatcherConfig:
|
|
session_id: str = ""
|
|
mode: EventMode = EventMode.PAPER
|
|
max_symbols: int = 25
|
|
min_bars: int = 10
|
|
ready_score_pct: float = 60.0
|
|
min_quality: float = 0.30
|
|
# Shared tradeable-population gate, applied BEFORE the per-tick evaluation.
|
|
# None = off (the historical behaviour every existing caller and test relies
|
|
# on). The live loop passes PopulationConfig.from_env() so the watcher stops
|
|
# scoring names the entry path can never accept: on 2026-09-10 the watcher
|
|
# wrote 191,458 criteria receipts across 57 symbols and 154,260 of them
|
|
# (80.6%) were for gaps outside the tradeable band — large caps like NOK, F
|
|
# and SOFI that sit inside the $2-20 price filter but can never pass the gap
|
|
# floor. Those two event types were 3.1 GB of a 7.6 GB events table.
|
|
population: "PopulationConfig | None" = None
|
|
|
|
|
|
@dataclass
|
|
class WatcherTickResult:
|
|
session_date: str
|
|
evaluated: int = 0
|
|
ready: list[str] = field(default_factory=list)
|
|
blocked: list[str] = field(default_factory=list)
|
|
discovered: list[str] = field(default_factory=list)
|
|
skipped: int = 0 # rejected by the population gate before evaluation
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
|
|
class Watcher:
|
|
"""Event-emitting intraday watcher."""
|
|
|
|
def __init__(
|
|
self,
|
|
store: EventStore,
|
|
provider: WatchlistProvider,
|
|
config: WatcherConfig | None = None,
|
|
catalyst_score_provider: Callable[[str], float | None] | None = None,
|
|
spread_provider: Callable[[str], float | None] | None = None,
|
|
):
|
|
self.store = store
|
|
self.provider = provider
|
|
self.config = config or WatcherConfig()
|
|
# optional symbol -> 0..1 catalyst score (or None). Injected so the pure
|
|
# evaluator stays DB-free; None provider == today's behavior exactly.
|
|
self.catalyst_score_provider = catalyst_score_provider
|
|
# optional symbol -> decision-time bid/ask spread fraction (or None).
|
|
# Observe-only: it is logged onto the evaluation/ready events but does NOT
|
|
# gate. None provider == today's behavior exactly (spread_pct logs as None,
|
|
# matching the always-null minute_bars.spread_pct column).
|
|
self.spread_provider = spread_provider
|
|
if not self.config.session_id:
|
|
self.config.session_id = f"session-{date.today().isoformat()}"
|
|
# symbol -> current state for this session
|
|
self._states: dict[str, str] = {}
|
|
|
|
# -- state helpers -------------------------------------------------
|
|
|
|
def _transition(self, symbol: str, new_state: str, reason: str | None) -> None:
|
|
previous = self._states.get(symbol)
|
|
if previous == new_state:
|
|
return
|
|
self._states[symbol] = new_state
|
|
self.store.emit(
|
|
SymbolStateChangedEvent(
|
|
timestamp=datetime.now(),
|
|
mode=self.config.mode,
|
|
correlation_id=self.config.session_id,
|
|
message=f"{symbol}: {previous or 'new'} -> {new_state}"
|
|
+ (f" ({reason})" if reason else ""),
|
|
symbol=symbol,
|
|
previous_state=previous,
|
|
new_state=new_state,
|
|
state_reason=reason,
|
|
payload={"symbol": symbol, "previous_state": previous,
|
|
"new_state": new_state, "state_reason": reason},
|
|
)
|
|
)
|
|
|
|
def _population_reject(self, candidate: WatchCandidate, bars) -> str | None:
|
|
"""Reject reason if this name is outside the tradeable population, else None.
|
|
|
|
Uses the STATIC OPEN gap (session open vs previous close) — the same
|
|
open-knowable quantity the live entry gate uses, and fixed for the day —
|
|
never the live price, which would make the verdict wander intraday.
|
|
Fail-open throughout: anything unreadable returns None so a data gap can
|
|
never silently bench a symbol.
|
|
"""
|
|
try:
|
|
# The SESSION open is the first REGULAR-HOURS bar's open. bars[0] is a
|
|
# PREMARKET bar before 09:30, which produces a fabricated gap: FTFT was
|
|
# benched 2026-09-14 as "gap -12.5%" when its true static open gap was
|
|
# +14.6%, and it never recovered because the frame keeps its premarket
|
|
# bars all session and _transition only fires on change.
|
|
if "is_regular_hours" in bars:
|
|
rth = bars[bars["is_regular_hours"] == True] # noqa: E712 — pandas mask
|
|
if len(rth) == 0:
|
|
return None # session has not opened — gap unknowable, fail open
|
|
open_px = float(rth["open"].iloc[0])
|
|
else:
|
|
# no session flag: the frame is already regular-hours only (tests,
|
|
# replay). Using bars[0] is correct there.
|
|
open_px = float(bars["open"].iloc[0])
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
prev_close = candidate.previous_close
|
|
gap = None
|
|
if prev_close:
|
|
try:
|
|
gap = (open_px - float(prev_close)) / float(prev_close)
|
|
except (TypeError, ValueError, ZeroDivisionError):
|
|
gap = None
|
|
volume = rvol = None
|
|
try:
|
|
volume = float(bars["volume"].sum())
|
|
adv = candidate.avg_daily_volume
|
|
if adv and float(adv) > 0:
|
|
rvol = volume / float(adv)
|
|
except Exception: # noqa: BLE001
|
|
volume = rvol = None
|
|
return population_reject(
|
|
self.config.population,
|
|
gap_pct=gap, price=open_px, volume=volume, rvol=rvol, as_of="open",
|
|
)
|
|
|
|
def _discover(self, candidate: WatchCandidate) -> None:
|
|
if candidate.symbol in self._states:
|
|
return
|
|
self.store.emit(
|
|
SymbolDiscoveredEvent(
|
|
timestamp=datetime.now(),
|
|
mode=self.config.mode,
|
|
correlation_id=self.config.session_id,
|
|
message=f"Discovered {candidate.symbol} via {candidate.source}",
|
|
symbol=candidate.symbol,
|
|
symbol_data={
|
|
"last_price": candidate.last_price,
|
|
"previous_close": candidate.previous_close,
|
|
"avg_daily_volume": candidate.avg_daily_volume,
|
|
"source": candidate.source,
|
|
},
|
|
payload={"symbol": candidate.symbol, "source": candidate.source},
|
|
)
|
|
)
|
|
self._transition(candidate.symbol, "watching", "added to watchlist")
|
|
|
|
# -- main loop body ------------------------------------------------
|
|
|
|
def tick(self, session_date: date | None = None) -> WatcherTickResult:
|
|
"""One evaluation pass over all candidates."""
|
|
session_date = session_date or date.today()
|
|
result = WatcherTickResult(session_date=session_date.isoformat())
|
|
|
|
try:
|
|
candidates = self.provider.get_candidates(session_date)
|
|
except Exception as exc: # provider failure should not kill the loop
|
|
logger.exception("watchlist provider failed")
|
|
result.errors.append(f"provider: {exc}")
|
|
return result
|
|
|
|
for candidate in candidates[: self.config.max_symbols]:
|
|
symbol = candidate.symbol
|
|
try:
|
|
if symbol not in self._states:
|
|
self._discover(candidate)
|
|
result.discovered.append(symbol)
|
|
|
|
bars = self.provider.get_bars(symbol, session_date)
|
|
if bars is None or len(bars) < self.config.min_bars:
|
|
self._transition(
|
|
symbol, "watching",
|
|
f"waiting for data ({0 if bars is None else len(bars)} bars)",
|
|
)
|
|
continue
|
|
|
|
# POPULATION PRE-FILTER — skip the expensive evaluation (and its
|
|
# two fat receipts) for a name the entry path could never accept.
|
|
# Recomputed every tick rather than cached: a >150% gapper starts
|
|
# "squeeze unverified" on thin volume and is admitted later once
|
|
# rvol confirms, so a cached rejection would bench it for the
|
|
# session. _transition only emits on CHANGE, so a rejected symbol
|
|
# costs exactly one event and then goes silent.
|
|
if self.config.population is not None:
|
|
reason = self._population_reject(candidate, bars)
|
|
if reason is not None:
|
|
self._transition(symbol, "out_of_population", reason)
|
|
result.skipped += 1
|
|
continue
|
|
|
|
# Evaluate the setup as of the latest bar's timestamp, not the
|
|
# wall clock. Minute bars are stored UTC-naive, but the entry
|
|
# cutoff reasons in US/Eastern wall-clock, so convert. This
|
|
# keeps replay and backfilled data honest (a 09:40 ET bar is
|
|
# judged against the 09:40 cutoff regardless of run time).
|
|
eval_time = None
|
|
try:
|
|
from datetime import timezone as _tz
|
|
from zoneinfo import ZoneInfo as _ZI
|
|
|
|
last_ts = pd.Timestamp(bars["timestamp"].iloc[-1]).to_pydatetime()
|
|
if last_ts.tzinfo is None:
|
|
last_ts = last_ts.replace(tzinfo=_tz.utc)
|
|
eval_time = last_ts.astimezone(_ZI("America/New_York")).replace(
|
|
tzinfo=None
|
|
)
|
|
except Exception: # noqa: BLE001 - fall back to now()
|
|
eval_time = None
|
|
|
|
# optional LLM catalyst score (None unless the blend is enabled
|
|
# AND a provider is injected) — never raises into the loop.
|
|
cat_score = None
|
|
if self.catalyst_score_provider is not None:
|
|
try:
|
|
cat_score = self.catalyst_score_provider(symbol)
|
|
except Exception: # noqa: BLE001
|
|
cat_score = None
|
|
|
|
evaluation = evaluate_setup(
|
|
bars,
|
|
previous_close=candidate.previous_close,
|
|
avg_daily_volume=candidate.avg_daily_volume,
|
|
ready_score_pct=self.config.ready_score_pct,
|
|
evaluation_time=eval_time,
|
|
min_bars=self.config.min_bars,
|
|
catalyst_score=cat_score,
|
|
)
|
|
result.evaluated += 1
|
|
|
|
# Decision-time bid/ask spread (observe-only). Guarded like the
|
|
# catalyst provider so a quote blip never kills the loop; None
|
|
# when no provider is injected or the quote is missing/bad.
|
|
spread_pct = None
|
|
if self.spread_provider is not None:
|
|
try:
|
|
spread_pct = self.spread_provider(symbol)
|
|
except Exception: # noqa: BLE001
|
|
spread_pct = None
|
|
|
|
self.store.emit(
|
|
CriteriaEvaluatedEvent(
|
|
timestamp=datetime.now(),
|
|
mode=self.config.mode,
|
|
correlation_id=self.config.session_id,
|
|
message=(
|
|
f"{symbol} criteria {evaluation.criteria_passed}/"
|
|
f"{evaluation.criteria_total} "
|
|
f"({evaluation.success_score_pct:.0f}%)"
|
|
),
|
|
symbol=symbol,
|
|
criteria_results={
|
|
"passed": evaluation.criteria_names_passed,
|
|
"failed": evaluation.criteria_names_failed,
|
|
"detail": evaluation.criteria_detail,
|
|
"gap_pct": evaluation.gap_pct,
|
|
"relative_volume": evaluation.relative_volume,
|
|
"spread_pct": spread_pct,
|
|
"status": evaluation.status,
|
|
"reason": evaluation.reason,
|
|
},
|
|
total_criteria=evaluation.criteria_total,
|
|
passed_criteria=evaluation.criteria_passed,
|
|
success_score_pct=evaluation.success_score_pct,
|
|
payload={
|
|
"symbol": symbol,
|
|
"success_score_pct": evaluation.success_score_pct,
|
|
"status": evaluation.status,
|
|
"reason": evaluation.reason,
|
|
"gap_pct": evaluation.gap_pct,
|
|
"relative_volume": evaluation.relative_volume,
|
|
"spread_pct": spread_pct,
|
|
},
|
|
)
|
|
)
|
|
|
|
if evaluation.status == "ready" and evaluation.setups:
|
|
result.ready.append(symbol)
|
|
if self._states.get(symbol) != "ready":
|
|
setup = evaluation.setups[0]
|
|
self._transition(symbol, "ready", setup["setup_type"])
|
|
signal_data = {**setup, "spread_pct": spread_pct}
|
|
self.store.emit(
|
|
SignalReadyEvent(
|
|
timestamp=datetime.now(),
|
|
mode=self.config.mode,
|
|
correlation_id=self.config.session_id,
|
|
message=(
|
|
f"{symbol} READY {setup['setup_type']} "
|
|
f"entry {setup['entry_price']} stop "
|
|
f"{setup['stop_loss_price']}"
|
|
),
|
|
symbol=symbol,
|
|
signal_type=setup["setup_type"],
|
|
confidence=setup["confidence"],
|
|
signal_data=signal_data,
|
|
payload={
|
|
"symbol": symbol,
|
|
"signal_type": setup["setup_type"],
|
|
"confidence": setup["confidence"],
|
|
"signal_data": signal_data,
|
|
},
|
|
)
|
|
)
|
|
else:
|
|
status = "blocked" if evaluation.status != "late" else "late"
|
|
result.blocked.append(symbol)
|
|
if self._states.get(symbol) == "ready":
|
|
# fell out of ready — allow re-signal later
|
|
self._transition(symbol, status, evaluation.reason)
|
|
elif self._states.get(symbol) != status:
|
|
self._transition(symbol, status, evaluation.reason)
|
|
self.store.emit(
|
|
SignalBlockedEvent(
|
|
timestamp=datetime.now(),
|
|
mode=self.config.mode,
|
|
correlation_id=self.config.session_id,
|
|
message=f"{symbol} blocked: {evaluation.reason}",
|
|
symbol=symbol,
|
|
blocking_reason=evaluation.reason or "unknown",
|
|
unmet_criteria=evaluation.criteria_names_failed,
|
|
payload={
|
|
"symbol": symbol,
|
|
"blocking_reason": evaluation.reason,
|
|
"unmet_criteria": evaluation.criteria_names_failed,
|
|
},
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
logger.exception("watcher tick failed for %s", symbol)
|
|
result.errors.append(f"{symbol}: {exc}")
|
|
|
|
return result
|