fez5587/momentum-system
1
0 Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
momentum-system/runtime/watcher.py
T
momentum devandClaude Fable 5.1 5f0625fb6f fix: watcher spends its 25-slot evaluation budget on in-population names first 
The watcher evaluates candidates[:max_symbols] each tick. The live provider puts
dashboard/scan-injected symbols first (they never expire) and ranks the rest by
bar count, so on 2026-10-08 fifteen injected intraday movers plus liquid names
filled all 25 slots and every in-population gapper stopped being scored at 09:43
while 50 already-rejected names were re-rejected every 8 s. Entries were not
affected (the trigger book arms independently and injects its armed names), but
the state board and the judge's reasoning trail went stale for the session.

Names already in out_of_population now go to the back of the list and rotate, so
they are still re-checked when budget remains (a squeeze can be re-admitted) but
can never crowd out a name that can still trade.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LXxioW7HA18AY1gqnzA2HN
2026-10-10 10:04:04 -05:00

409 lines
19 KiB
Python
Raw Permalink Blame History

"""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"
# spike-robust baseline for squeeze verification (see _population_reject)
median_daily_volume: float | None = None
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] = {}
self._tick_no = 0 # rotates which out-of-population names get re-checked
# -- 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())
# squeeze verification takes the stronger of the mean- and
# median-baseline readings (same units): a prior spike day in the
# 20d window must not bench a real squeeze (KITT 2026-09-25).
readings = [volume / float(b) for b in (candidate.avg_daily_volume,
candidate.median_daily_volume)
if b and float(b) > 0]
rvol = max(readings) if readings else None
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
# The evaluation budget is max_symbols candidates per tick. Spend it on
# names that can still trade: on 2026-10-08 the live list was 15 scan-
# injected intraday movers (placed first by the provider, never expiring)
# plus research rows ranked by bar count, so every in-population gapper
# fell past position 25 and stopped being scored at 09:43 while 50
# already-rejected names were re-rejected every 8 s. Rejected names go to
# the back and rotate, so a squeeze can still be re-admitted when there is
# budget left, but they never crowd out a live candidate.
front = [c for c in candidates if self._states.get(c.symbol) != "out_of_population"]
back = [c for c in candidates if self._states.get(c.symbol) == "out_of_population"]
if back:
k = self._tick_no % len(back)
back = back[k:] + back[:k]
self._tick_no += 1
for candidate in (front + back)[: 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
Reference in new issue View Git Blame Copy Permalink