momentum devandClaude Opus 4.8
44ff36a669
The 2026-07-08 audit (owner: "figure out if there is some kind of logic or audit issue") found BOTH entry paths feeding the population gate the WRONG gap. The fast path used the trigger book's dynamic gap (latest price vs prev close), so intraday rallies masqueraded as gappers: TVRD (6.0% true gap), SKIN (2.2%, filled at $0.98), JLHL (4.3%) all traded outside the validated population. The slow path used max(gap, intraday) with "open" taken from the first lookback-window bar, and had no upper band at all — dead-band names SEER (38.6%) and IOTR (40.7%) entered. Fixes: candidates carry gap_open_pct (session open vs prev close); trigger eligibility + submit_breakout_now use it exclusively once the session is open; _population_reject now implements the full validated band (10-35% OR volume-verified >=150% squeeze, $1-20 price band) for both paths; evaluator gap uses the session open; signals carry relative_volume for squeeze verification. Plus momentum-testpg.service so the test-isolation Postgres survives reboots. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1189 lines
55 KiB
Python
1189 lines
55 KiB
Python
"""Live paper-trading orchestrator (Milestone 4 end-to-end).
|
|
|
|
This is the piece that was missing: a single loop tying together
|
|
|
|
1. ingest — latest 1-minute Alpaca bars into the market schema (Postgres)
|
|
(plus a daily-bar backfill on startup so gap %% / RVOL work)
|
|
2. watch — Watcher.tick over the ingested bars, emitting signal_ready
|
|
3. sync — Alpaca paper account / positions / orders into the event DB
|
|
4. execute — TradingExecutionService.tick: approval requests, and order
|
|
submission (auto or via dashboard approval)
|
|
5. dashboard — optional embedded API + UI at http://127.0.0.1:8010
|
|
|
|
Usage
|
|
-----
|
|
python run_live_paper.py # continuous loop
|
|
python run_live_paper.py --once # single pass, then exit
|
|
python run_live_paper.py --symbols AAPL,SNDL --auto-approve
|
|
python run_live_paper.py --no-dashboard
|
|
|
|
Without ALPACA_API_KEY / ALPACA_SECRET_KEY the loop runs in dry mode:
|
|
network steps are skipped with clear messages, but watcher + dashboard
|
|
still work against whatever data already exists.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
from alpaca_paper.client import AlpacaPaperClient
|
|
from alpaca_paper.execution import AlpacaPaperExecutor
|
|
from alpaca_paper.settings import AlpacaPaperSettings
|
|
from alpaca_paper.sync import AlpacaPaperSync
|
|
from api.main import DashboardState, serve_in_background
|
|
from research.ingestion.market_data import (
|
|
classify_session,
|
|
discover_active_symbols,
|
|
ingest_daily_history,
|
|
ingest_live_minute_bars,
|
|
)
|
|
from research.ingestion.scheduler import Scheduler
|
|
from research.ingestion.watcher_task import ResearchWatchlistProvider
|
|
from research.ingestion.discovery import run_discovery, screen_universe
|
|
from research.ingestion.alpaca_news import ingest_alpaca_news
|
|
from research.ingestion.news_enrichment import enrich_recent_news
|
|
from config import OllamaConfig
|
|
from research.multi_schema import open_research_db
|
|
from runtime.exit_manager import LiveExitManager
|
|
from runtime.halt_guard import is_halt_gap
|
|
from runtime.market_calendar import (
|
|
days_to_next_session,
|
|
eod_flatten_status,
|
|
is_regular_hours,
|
|
)
|
|
from runtime.triggers import ArmedTriggerBook
|
|
from runtime.watcher import Watcher, WatcherConfig
|
|
from strategy.exits import ExitConfig
|
|
from storage.event_schema import EventMode, ModuleTickEvent
|
|
from storage.event_store import EventStore
|
|
from trading_execution import ExecutionSettings, TradingExecutionService
|
|
from trading_mode import TradingModeSettings
|
|
|
|
# Fallback watchlist used ONLY if the sub-$20 screener is unavailable. Liquid
|
|
# small-caps that usually sit in the $1-20 band — NOT mega-caps (which the
|
|
# watcher's price band would just filter out, leaving nothing to trade).
|
|
DEFAULT_SYMBOLS = [
|
|
"SOFI", "GRAB", "NU", "AAL", "PLUG", "RIOT",
|
|
"MARA", "SNDL", "ACHR", "NOK", "SOUN", "BBAI",
|
|
]
|
|
|
|
|
|
def _flag(name: str, default: str = "1") -> bool:
|
|
return os.environ.get(name, default).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _now_session_date():
|
|
session_date, _, _, _ = classify_session(datetime.now(timezone.utc))
|
|
return session_date
|
|
|
|
|
|
def _load_learned() -> dict:
|
|
"""Nightly-tuned params from data/learned_params.json (empty if absent)."""
|
|
path = os.path.join(os.environ.get("DATA_DIR", "./data"), "learned_params.json")
|
|
try:
|
|
with open(path) as fh:
|
|
return json.load(fh)
|
|
except Exception: # noqa: BLE001
|
|
return {}
|
|
|
|
|
|
def _db_target_desc() -> str:
|
|
"""Human-readable Postgres target for the boot log (password masked)."""
|
|
url = os.environ.get("DATABASE_URL")
|
|
if url:
|
|
if "://" in url and "@" in url:
|
|
scheme, rest = url.split("://", 1)
|
|
creds, host = rest.split("@", 1)
|
|
user = creds.split(":", 1)[0]
|
|
return f"{scheme}://{user}:***@{host}"
|
|
return url
|
|
host = os.environ.get("PGHOST", "localhost")
|
|
port = os.environ.get("PGPORT", "5432")
|
|
db = os.environ.get("PGDATABASE", "momentum")
|
|
user = os.environ.get("PGUSER", "")
|
|
return f"postgresql://{user}@{host}:{port}/{db}"
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Momentum live paper-trading loop")
|
|
parser.add_argument("--once", action="store_true", help="run one pass and exit")
|
|
parser.add_argument(
|
|
"--symbols",
|
|
default=os.environ.get("WATCHER_SYMBOLS", ""),
|
|
help="comma-separated symbols to watch (default: WATCHER_SYMBOLS env or built-ins)",
|
|
)
|
|
parser.add_argument(
|
|
"--auto-approve",
|
|
action="store_true",
|
|
help="auto-approve and submit ready signals (otherwise approve from the dashboard)",
|
|
)
|
|
parser.add_argument(
|
|
"--discover",
|
|
action="store_true",
|
|
help="also pull symbols from Alpaca's most-actives screener",
|
|
)
|
|
parser.add_argument("--no-dashboard", action="store_true", help="skip the embedded dashboard")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def build_runtime(args: argparse.Namespace) -> dict:
|
|
"""Wire up every component; safe without API keys."""
|
|
load_dotenv()
|
|
|
|
explicit_symbols = [s.strip().upper() for s in args.symbols.split(",") if s.strip()]
|
|
|
|
alpaca_settings = AlpacaPaperSettings.from_env()
|
|
has_keys = alpaca_settings.is_configured
|
|
client = AlpacaPaperClient(alpaca_settings) if has_keys else None
|
|
|
|
# Datastore is Postgres (DATABASE_URL); WATCHER_EVENT_DB_PATH is a legacy
|
|
# fallback the connection layer ignores.
|
|
event_db = os.environ.get("WATCHER_EVENT_DB_PATH", "momentum")
|
|
store = EventStore(event_db)
|
|
research_con = open_research_db("market")
|
|
# NBBO quote-size snapshots for tracked/held names — the free proxy for order-book
|
|
# depth imbalance (the trader's "big seller at $6" read). Recording-only; a shadow
|
|
# study consumes it later. See step_quotes.
|
|
research_con.execute(
|
|
"""CREATE TABLE IF NOT EXISTS quote_snapshots (
|
|
symbol TEXT NOT NULL,
|
|
ts TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
bid_price DOUBLE PRECISION, bid_size DOUBLE PRECISION,
|
|
ask_price DOUBLE PRECISION, ask_size DOUBLE PRECISION)""")
|
|
research_con.execute(
|
|
"CREATE INDEX IF NOT EXISTS quote_snaps_sym_ts ON quote_snapshots(symbol, ts)")
|
|
|
|
session_id = f"paper-{_now_session_date().isoformat()}-{uuid.uuid4().hex[:6]}"
|
|
|
|
# Watchlist resolution. Explicit --symbols/WATCHER_SYMBOLS win; otherwise
|
|
# discover the sub-$20 most-actives universe via the screener (the strategy
|
|
# targets $1-20 small-caps, not the hardcoded mega-cap fallback). Falls back
|
|
# to DEFAULT_SYMBOLS only if the screener is unavailable.
|
|
price_min = float(os.environ.get("WATCHER_PRICE_MIN", "1"))
|
|
price_max = float(os.environ.get("WATCHER_PRICE_MAX", "20"))
|
|
discover_top = int(os.environ.get("DISCOVER_TOP", "20"))
|
|
if explicit_symbols:
|
|
symbols = explicit_symbols
|
|
symbols_source = "explicit"
|
|
elif client is not None and _flag("DISCOVER_ON_START", "1"):
|
|
discovered = screen_universe(
|
|
client, price_min=price_min, price_max=price_max, top=discover_top
|
|
)
|
|
symbols = discovered or list(DEFAULT_SYMBOLS)
|
|
symbols_source = "screener" if discovered else "fallback"
|
|
else:
|
|
symbols = list(DEFAULT_SYMBOLS)
|
|
symbols_source = "fallback"
|
|
|
|
store.emit(
|
|
ModuleTickEvent(
|
|
timestamp=datetime.now(),
|
|
mode=EventMode.PAPER,
|
|
correlation_id=session_id,
|
|
message=f"watchlist resolved via {symbols_source}: {len(symbols)} symbols",
|
|
module="discovery",
|
|
stage="boot",
|
|
duration_ms=0.0,
|
|
input_count=0,
|
|
output_count=len(symbols),
|
|
metrics={"source": symbols_source, "symbols": symbols},
|
|
errors=(
|
|
[]
|
|
if symbols_source != "fallback"
|
|
else [{"error": "screener returned 0 names at boot — using fallback "
|
|
"watchlist; check Alpaca screener / data plan"}]
|
|
),
|
|
)
|
|
)
|
|
|
|
_learned = _load_learned()
|
|
if _learned:
|
|
print(f"[boot] learned params: ready_score={_learned.get('ready_score_pct')} "
|
|
f"min_bars={_learned.get('min_bars')} setups={_learned.get('setups')} "
|
|
f"(tuned {str(_learned.get('as_of', ''))[:10]} over {_learned.get('sessions')} "
|
|
f"sessions; backtest pnl {_learned.get('pnl')}, entry@{_learned.get('entry_min')}min)")
|
|
|
|
ollama_cfg = OllamaConfig.from_env()
|
|
|
|
# Catalyst advisory provider, TTL-cached so a pass doesn't re-query per symbol;
|
|
# returns None on any fault so neither the (gated) dilution veto nor the score
|
|
# blend ever block/skew on a DB blip. Shared by the watcher AND the execution
|
|
# service (one cache). catalyst_score_provider maps the advisory -> a 0..1 score
|
|
# OUTSIDE the pure evaluator, and is only handed to the watcher when the score
|
|
# blend is explicitly enabled (so Phase 1 advisory never changes scoring).
|
|
_catalyst_cache: dict = {"map": {}, "at": 0.0, "con": None}
|
|
_catalyst_lock = threading.Lock()
|
|
_CATALYST_TTL = 30.0
|
|
|
|
def catalyst_provider(symbol: str):
|
|
import time as _time
|
|
from research.ingestion.news_enrichment import catalyst_map
|
|
now = _time.monotonic()
|
|
# Called from the TRIGGER thread (submit_breakout_now veto), the scheduler
|
|
# thread (approvals), AND the watcher — so it must NOT touch the shared
|
|
# research_con. Refresh under a lock using a DEDICATED connection (opened
|
|
# lazily, only inside the lock => single-threaded use); the 30s TTL keeps
|
|
# DB hits rare, and the read below is a lock-free atomic dict get.
|
|
if (now - _catalyst_cache["at"]) >= _CATALYST_TTL:
|
|
with _catalyst_lock:
|
|
if (_time.monotonic() - _catalyst_cache["at"]) >= _CATALYST_TTL:
|
|
try:
|
|
if _catalyst_cache["con"] is None:
|
|
_catalyst_cache["con"] = open_research_db("market")
|
|
_catalyst_cache["map"] = catalyst_map(_catalyst_cache["con"])
|
|
except Exception: # noqa: BLE001
|
|
_catalyst_cache["map"] = {}
|
|
_catalyst_cache["at"] = _time.monotonic()
|
|
return _catalyst_cache["map"].get((symbol or "").upper())
|
|
|
|
def catalyst_score_provider(symbol: str):
|
|
from research.ingestion.news_enrichment import catalyst_score
|
|
return catalyst_score(catalyst_provider(symbol))
|
|
|
|
# Decision-time bid/ask spread provider (observe-only). One per-symbol quote,
|
|
# TTL-cached so a watch pass doesn't re-fetch, mapped to spread_pct. Returns
|
|
# None on any fault / missing quote / when disabled, so the watcher logs the
|
|
# metric without ever blocking or breaking on a quote blip. Default ON when a
|
|
# client exists; set SPREAD_PCT_ENABLED=0 to silence the extra quote calls.
|
|
_spread_cache: dict[str, tuple[float, float | None]] = {}
|
|
_SPREAD_TTL = 15.0
|
|
|
|
def _spread_enabled() -> bool:
|
|
return os.environ.get("SPREAD_PCT_ENABLED", "1").strip().lower() in {
|
|
"1", "true", "yes", "on"}
|
|
|
|
def spread_provider(symbol: str):
|
|
if client is None or not _spread_enabled():
|
|
return None
|
|
import time as _time
|
|
from strategy.evaluation.liquidity import compute_spread_pct
|
|
now = _time.monotonic()
|
|
hit = _spread_cache.get(symbol)
|
|
if hit is not None and (now - hit[0]) < _SPREAD_TTL:
|
|
return hit[1]
|
|
val = None
|
|
try:
|
|
quote = (client.get_latest_quotes([symbol]) or {}).get(symbol) or {}
|
|
val = compute_spread_pct(quote.get("bp"), quote.get("ap"))
|
|
except Exception: # noqa: BLE001
|
|
val = None
|
|
_spread_cache[symbol] = (now, val)
|
|
return val
|
|
|
|
watcher = Watcher(
|
|
store,
|
|
ResearchWatchlistProvider(
|
|
research_con,
|
|
limit=int(os.environ.get("WATCHER_MAX_SYMBOLS", "25")),
|
|
),
|
|
WatcherConfig(
|
|
session_id=session_id,
|
|
mode=EventMode.PAPER,
|
|
max_symbols=int(os.environ.get("WATCHER_MAX_SYMBOLS", "25")),
|
|
# Defaults come from the nightly self-tuner (learned_params.json);
|
|
# an explicit env var still wins.
|
|
min_bars=int(os.environ.get("WATCHER_MIN_BARS", str(_learned.get("min_bars", 10)))),
|
|
ready_score_pct=float(os.environ.get(
|
|
"WATCHER_READY_SCORE", str(_learned.get("ready_score_pct", 60.0)))),
|
|
min_quality=float(os.environ.get("WATCHER_MIN_QUALITY", "0.30")),
|
|
),
|
|
catalyst_score_provider=(
|
|
catalyst_score_provider if ollama_cfg.catalyst_score_enabled else None
|
|
),
|
|
spread_provider=spread_provider,
|
|
)
|
|
|
|
exec_settings = ExecutionSettings.from_env()
|
|
if args.auto_approve:
|
|
exec_settings.auto_approve = True
|
|
executor = AlpacaPaperExecutor(store, client=client, session_id=session_id) if client else None
|
|
|
|
# Live last-trade price source for entry invalidation — tick-by-tick, not
|
|
# bar close. Pulls the most recent trade from Alpaca's data API on demand;
|
|
# a tiny TTL cache collapses redundant lookups within a single guard pass.
|
|
# On any failure it returns None so the price-break check is simply skipped
|
|
# for that pass (the wall-clock timeout still protects the entry) rather
|
|
# than silently falling back to a stale bar close.
|
|
_price_cache: dict[str, tuple[float, float]] = {}
|
|
_PRICE_TTL = 2.0
|
|
|
|
def latest_price(symbol: str):
|
|
if client is None:
|
|
return None
|
|
import time as _time
|
|
|
|
cached = _price_cache.get(symbol)
|
|
now = _time.monotonic()
|
|
if cached and (now - cached[1]) < _PRICE_TTL:
|
|
return cached[0]
|
|
try:
|
|
trades = client.get_latest_trades([symbol])
|
|
trade = trades.get(symbol) or trades.get(symbol.upper())
|
|
price = float(trade["p"]) if trade and trade.get("p") is not None else None
|
|
except Exception:
|
|
price = None
|
|
if price is not None:
|
|
_price_cache[symbol] = (price, now)
|
|
return price
|
|
|
|
execution = (
|
|
TradingExecutionService(
|
|
store,
|
|
executor=executor,
|
|
settings=exec_settings,
|
|
trading_mode=TradingModeSettings.from_env(),
|
|
session_id=session_id,
|
|
price_provider=latest_price,
|
|
catalyst_provider=catalyst_provider,
|
|
)
|
|
if executor
|
|
else None
|
|
)
|
|
sync = AlpacaPaperSync(store, client=client, session_id=session_id) if client else None
|
|
|
|
# Fast armed-trigger book: keep the N most-promising gappers ready with a
|
|
# pre-computed opening-range trigger/stop so step_trigger can fire the
|
|
# instant live price crosses (instead of waiting for a bar close + cadence).
|
|
book = ArmedTriggerBook(
|
|
max_armed=int(os.environ.get("TRIGGER_MAX_ARMED", "6")),
|
|
gap_min=float(os.environ.get("TRIGGER_GAP_MIN", "3.0")),
|
|
gap_max=float(os.environ.get("TRIGGER_GAP_MAX", "1e9")),
|
|
gap_squeeze_min=float(os.environ.get("TRIGGER_GAP_SQUEEZE_MIN", "0")),
|
|
gap_squeeze_rvol=float(os.environ.get("TRIGGER_GAP_SQUEEZE_RVOL", "5.0")),
|
|
rvol_min=float(os.environ.get("TRIGGER_RVOL_MIN", "2.0")),
|
|
min_range_pct=float(os.environ.get("TRIGGER_MIN_RANGE_PCT", "0.004")),
|
|
min_dollar_vol=float(os.environ.get("TRIGGER_MIN_DOLLAR_VOL", "0")),
|
|
max_price_age_s=float(os.environ.get("TRIGGER_PRICE_MAX_AGE", "12")),
|
|
)
|
|
|
|
return {
|
|
"symbols": symbols,
|
|
"symbols_source": symbols_source,
|
|
# per-symbol IEX->consolidated volume correction (research.ingestion.volume_ratio):
|
|
# measured each morning from the delayed free SIP feed, applied at ingest so live
|
|
# RVOL/liquidity signals are consolidated-scale like their (nightly-SIP) baselines
|
|
"vol_ratios": {},
|
|
"client": client,
|
|
"has_keys": has_keys,
|
|
"store": store,
|
|
"research_con": research_con,
|
|
"watcher": watcher,
|
|
"execution": execution,
|
|
"sync": sync,
|
|
"book": book,
|
|
"latest_price": latest_price,
|
|
"catalyst_provider": catalyst_provider,
|
|
"ollama_cfg": ollama_cfg,
|
|
"session_id": session_id,
|
|
"event_db": event_db,
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
rt = build_runtime(args)
|
|
ollama_cfg = rt["ollama_cfg"]
|
|
symbols: list[str] = rt["symbols"]
|
|
client = rt["client"]
|
|
|
|
print(f"[boot] session={rt['session_id']} db={_db_target_desc()}")
|
|
print(f"[boot] watching ({rt['symbols_source']}): {', '.join(symbols)}")
|
|
if not rt["has_keys"]:
|
|
print("[boot] no ALPACA_API_KEY/SECRET — dry mode: ingestion, sync and "
|
|
"order submission are skipped; watcher runs on existing data")
|
|
|
|
# dashboard
|
|
server = None
|
|
if not args.no_dashboard and _flag("DASHBOARD_ENABLED", "1"):
|
|
host = os.environ.get("DASHBOARD_HOST", "127.0.0.1")
|
|
port = int(os.environ.get("DASHBOARD_PORT", "8010"))
|
|
|
|
def inject_symbol(symbol: str) -> None:
|
|
# add to the watcher's provider so it's evaluated next tick, and
|
|
# to the live symbol list so the next ingest pass pulls its bars
|
|
provider = rt["watcher"].provider
|
|
if hasattr(provider, "add_symbol"):
|
|
provider.add_symbol(symbol)
|
|
if symbol not in symbols:
|
|
symbols.append(symbol)
|
|
|
|
state = DashboardState(
|
|
rt["event_db"],
|
|
execution_service=rt["execution"],
|
|
execution_mode=TradingModeSettings.from_env().execution_mode,
|
|
research_con=rt["research_con"],
|
|
watch_inject=inject_symbol,
|
|
)
|
|
try:
|
|
server, _ = serve_in_background(state, host, port)
|
|
print(f"[boot] dashboard at http://{host}:{port}")
|
|
except OSError as exc:
|
|
print(f"[boot] dashboard disabled ({exc})")
|
|
|
|
# startup: daily history backfill so gap%/RVOL have a baseline
|
|
if client:
|
|
if args.discover:
|
|
discovered = discover_active_symbols(
|
|
client,
|
|
top=10,
|
|
price_min=float(os.environ.get("WATCHER_PRICE_MIN", "1")),
|
|
price_max=float(os.environ.get("WATCHER_PRICE_MAX", "20")),
|
|
)
|
|
if discovered:
|
|
print(f"[boot] screener added: {', '.join(discovered)}")
|
|
symbols = sorted(set(symbols) | set(discovered))
|
|
else:
|
|
print("[boot] screener unavailable — using static symbol list")
|
|
daily = ingest_daily_history(rt["research_con"], client, symbols, days=30)
|
|
print(f"[boot] daily backfill: {daily.daily_rows} rows"
|
|
+ (f" (errors: {daily.errors})" if daily.errors else ""))
|
|
|
|
lookback = int(os.environ.get("LIVE_BARS_LOOKBACK_MINUTES", "240"))
|
|
refresh_lb = int(os.environ.get("LIVE_BARS_REFRESH_MINUTES", "20"))
|
|
_ingest_first = {"v": True}
|
|
# freshness watchdog: warn when the newest bar is older than this during RTH.
|
|
_ingest_stale_secs = float(os.environ.get("INGEST_STALE_SECONDS", "180"))
|
|
# block arming on stale data? OFF by default — observability first; a
|
|
# mis-tuned threshold must not silently halt trading.
|
|
_ingest_stale_block = os.environ.get("INGEST_STALE_BLOCK", "0").strip() in {"1", "true", "yes"}
|
|
_data_stale = {"v": False}
|
|
|
|
def _first_pass_lookback() -> int:
|
|
# The first ingest must cover the whole session-so-far so the
|
|
# opening-range bars land in the DB (even on a mid-session restart);
|
|
# after that we only pull the last few minutes — closed bars never
|
|
# change, and re-fetching a 4-hour window for 80 symbols every minute
|
|
# was the ~46s step that froze the loop.
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
now_et = datetime.now(ZoneInfo("America/New_York"))
|
|
open_et = now_et.replace(hour=9, minute=30, second=0, microsecond=0)
|
|
since_open = max(0.0, (now_et - open_et).total_seconds() / 60.0)
|
|
except Exception: # noqa: BLE001
|
|
since_open = 0.0
|
|
return int(max(lookback, since_open + 15))
|
|
|
|
def _freshest_bar_age():
|
|
"""Seconds since the newest stored minute bar (UTC), or None. The bar
|
|
timestamp is its START, and IEX lags ~1 bar, so ~90-120s is NORMAL during
|
|
RTH — only flag well beyond that."""
|
|
from datetime import timezone
|
|
try:
|
|
row = rt["research_con"].execute(
|
|
"SELECT max(timestamp) FROM minute_bars WHERE session_date = ?",
|
|
[_now_session_date()]).fetchone()
|
|
if not row or not row[0]:
|
|
return None
|
|
ts = row[0]
|
|
if isinstance(ts, str):
|
|
ts = datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
|
if ts.tzinfo is None:
|
|
ts = ts.replace(tzinfo=timezone.utc)
|
|
return (datetime.now(timezone.utc) - ts).total_seconds()
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
def _is_rth():
|
|
# holiday/early-close aware (was weekday<5 only, which traded holidays)
|
|
return is_regular_hours()
|
|
|
|
def _symbol_bar_age(symbol):
|
|
"""Seconds since this symbol's newest minute bar today (UTC), or None."""
|
|
from datetime import timezone
|
|
try:
|
|
row = rt["research_con"].execute(
|
|
"SELECT max(timestamp) FROM minute_bars "
|
|
"WHERE symbol = ? AND session_date = ?",
|
|
[symbol, _now_session_date()]).fetchone()
|
|
if not row or not row[0]:
|
|
return None
|
|
ts = row[0]
|
|
if isinstance(ts, str):
|
|
ts = datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
|
if ts.tzinfo is None:
|
|
ts = ts.replace(tzinfo=timezone.utc)
|
|
return (datetime.now(timezone.utc) - ts).total_seconds()
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
def _recently_halted(symbol):
|
|
"""Halt heuristic for the entry guard: during RTH this symbol has gone
|
|
silent (freshest bar older than the configured gap) WHILE ingestion is
|
|
otherwise healthy (some name printed recently). For the liquid gappers we
|
|
trade, a multi-minute silence is a trading halt, not thin coverage — and
|
|
entering a halted name is the "+100% then halts down" trap. Conservative
|
|
on purpose: a false positive just skips one entry (it retries next tick)."""
|
|
ex = rt.get("execution")
|
|
if ex is None or not ex.settings.halt_guard_enabled or not _is_rth():
|
|
return False
|
|
gap_s = ex.settings.halt_max_bar_gap_min * 60.0
|
|
# is_halt_gap also gates the global-ingestion-health check, so a feed-wide
|
|
# lag can't flag the whole book (pure + unit-tested in runtime.halt_guard)
|
|
return is_halt_gap(_symbol_bar_age(symbol), _freshest_bar_age(), gap_s)
|
|
|
|
def step_vratio():
|
|
"""Measure today's per-name IEX->SIP volume capture ratio from the delayed
|
|
(free) SIP feed and keep it for ingest scaling. Merge, never clear: a name
|
|
whose sample got too thin keeps its last good ratio for the session.
|
|
Self-disables when the live feed is already SIP (paid plan) — the estimator
|
|
is the free-tier fallback, and scaling consolidated volume would double-count
|
|
(ingest_live_minute_bars refuses too; this just saves the API calls)."""
|
|
if not client:
|
|
return "skipped (no keys)"
|
|
if getattr(client.settings, "feed", "iex") != "iex":
|
|
return "off (live feed is SIP)"
|
|
from research.ingestion.volume_ratio import compute_ratios, persist_ratios
|
|
fresh = compute_ratios(client, list(symbols))
|
|
if fresh:
|
|
rt["vol_ratios"].update(fresh)
|
|
persist_ratios(rt["research_con"], fresh)
|
|
known = rt["vol_ratios"]
|
|
if not known:
|
|
return "no ratios yet"
|
|
med = sorted(known.values())[len(known) // 2]
|
|
return f"{len(fresh)} fresh / {len(known)} known, median x{med:.0f}"
|
|
|
|
def step_ingest():
|
|
if not client:
|
|
return "skipped (no keys)"
|
|
if _ingest_first["v"]:
|
|
lb = _first_pass_lookback()
|
|
_ingest_first["v"] = False
|
|
else:
|
|
lb = refresh_lb
|
|
res = ingest_live_minute_bars(rt["research_con"], client, symbols, lookback_minutes=lb,
|
|
volume_scale=rt["vol_ratios"])
|
|
# Freshness watchdog: 45% of passes historically returned 0 rows with NO
|
|
# error — indistinguishable from a feed outage. Make staleness VISIBLE
|
|
# (emit + display); hard-block arming only if explicitly enabled, so a
|
|
# mis-tuned threshold can't silently halt trading.
|
|
age = _freshest_bar_age()
|
|
stale = bool(_is_rth() and age is not None and age > _ingest_stale_secs)
|
|
_data_stale["v"] = stale
|
|
if stale:
|
|
rt["store"].emit(ModuleTickEvent(
|
|
timestamp=datetime.now(), mode=EventMode.PAPER,
|
|
correlation_id=rt["session_id"],
|
|
message=f"INGEST STALE: freshest bar {age:.0f}s old (> {_ingest_stale_secs:.0f}s)",
|
|
module="ingestion", stage="stale", duration_ms=0.0,
|
|
input_count=len(symbols), output_count=res.minute_rows,
|
|
metrics={"bar_age_s": age}, errors=[{"error": "stale_data"}]))
|
|
rt["store"].emit(
|
|
ModuleTickEvent(
|
|
timestamp=datetime.now(),
|
|
mode=EventMode.PAPER,
|
|
correlation_id=rt["session_id"],
|
|
message=f"ingested {res.minute_rows} minute rows across {len(res.symbols)} symbols",
|
|
module="ingestion",
|
|
stage="completed",
|
|
duration_ms=0.0,
|
|
input_count=len(symbols),
|
|
output_count=res.minute_rows,
|
|
metrics={"per_symbol": res.per_symbol, "symbols_with_data": res.symbols},
|
|
errors=[{"error": e} for e in res.errors],
|
|
)
|
|
)
|
|
note = (f" STALE({age:.0f}s)" if stale else "")
|
|
return (f"{res.minute_rows} rows (lookback {lb}m){note}"
|
|
+ (f", errors={res.errors}" if res.errors else ""))
|
|
|
|
def step_watch():
|
|
res = rt["watcher"].tick(_now_session_date())
|
|
return (f"evaluated={res.evaluated} ready={res.ready} blocked={len(res.blocked)}"
|
|
+ (f" errors={res.errors}" if res.errors else ""))
|
|
|
|
def step_sync():
|
|
if not rt["sync"]:
|
|
return "skipped (no keys)"
|
|
res = rt["sync"].sync_all()
|
|
failed = [k for k, v in res.items() if v is None]
|
|
return "ok" if not failed else f"DEGRADED ({', '.join(failed)})"
|
|
|
|
def step_execute():
|
|
if not rt["execution"]:
|
|
return "skipped (no keys)"
|
|
res = rt["execution"].tick()
|
|
backed = len(res.get("backed_out") or [])
|
|
# `auto` is the raw approve_order() results: a REAL broker submission carries a
|
|
# broker_order_id; rejections (e.g. the daily-entry cap) are tagged rejected=True.
|
|
# Count only real submissions as executed, and surface rejections explicitly so a
|
|
# tripped cap is visible instead of masquerading as auto_executed=1.
|
|
auto = res.get("auto_executed") or []
|
|
executed = [r for r in auto if r.get("broker_order_id")]
|
|
rejected = [r for r in auto if r.get("rejected")]
|
|
line = (f"approvals_requested={res['approvals_requested']} "
|
|
f"auto_executed={len(executed)}")
|
|
if rejected:
|
|
hist: dict[str, int] = {}
|
|
for r in rejected:
|
|
k = r.get("reason") or "?"
|
|
hist[k] = hist.get(k, 0) + 1
|
|
reasons = ", ".join(f"{k}x{v}" for k, v in sorted(hist.items()))
|
|
line += f" auto_rejected={len(rejected)} ({reasons})"
|
|
if backed:
|
|
syms = ", ".join(b["symbol"] for b in res["backed_out"])
|
|
line += f" backed_out={backed} ({syms})"
|
|
return line
|
|
|
|
def step_guard():
|
|
# Fast invalidation watch: re-check armed (unfilled) entries against the
|
|
# live last-trade price and the wall-clock timeout. Runs far more often
|
|
# than `execute` so a break of the entry trigger is caught tick-by-tick
|
|
# instead of waiting for the next execute pass.
|
|
if not rt["execution"]:
|
|
return None
|
|
backed = rt["execution"].expire_stale_entries()
|
|
if not backed:
|
|
return None
|
|
return "backed_out " + ", ".join(
|
|
f"{b['symbol']} ({b['reason']})" for b in backed
|
|
)
|
|
|
|
def step_discover():
|
|
if not client:
|
|
return "skipped (no keys)"
|
|
res = run_discovery(
|
|
rt["store"], rt["research_con"], client, _now_session_date(),
|
|
mode=EventMode.PAPER, correlation_id=rt["session_id"],
|
|
price_min=float(os.environ.get("WATCHER_PRICE_MIN", "1")),
|
|
price_max=float(os.environ.get("WATCHER_PRICE_MAX", "20")),
|
|
top=int(os.environ.get("DISCOVER_TOP", "20")),
|
|
)
|
|
# fold newly-screened names into the live watchlist so the next ingest
|
|
# pulls their bars and the watcher evaluates them
|
|
added = [s for s in res.universe if s not in symbols]
|
|
symbols.extend(added)
|
|
# Keep the watchlist BOUNDED so it rotates toward current movers instead
|
|
# of growing all session. (Ingestion is now batched so an over-cap list
|
|
# no longer 400s, but an unbounded list is still wasteful and stale.)
|
|
max_uni = int(os.environ.get("WATCHER_MAX_UNIVERSE", "80"))
|
|
if len(symbols) > max_uni:
|
|
del symbols[: len(symbols) - max_uni]
|
|
return (f"universe={len(res.universe)} gappers={len(res.gappers)}"
|
|
+ (f" +{len(added)} new" if added else ""))
|
|
|
|
# --- fast armed-trigger path -------------------------------------------
|
|
book = rt["book"]
|
|
price_min_v = float(os.environ.get("WATCHER_PRICE_MIN", "1"))
|
|
price_max_v = float(os.environ.get("WATCHER_PRICE_MAX", "20"))
|
|
orb_bars_v = int(os.environ.get("ORB_BARS", "5"))
|
|
stop_cushion_v = float(os.environ.get("TRIGGER_STOP_CUSHION", "0.005"))
|
|
|
|
def step_arm():
|
|
"""Rank the most-promising gappers and pre-compute each one's opening-
|
|
range breakout trigger/stop, so step_trigger can fire on a live cross."""
|
|
from research import query as rq
|
|
from research.ingestion.discovery import recent_news_map
|
|
from research.ingestion.signals import scan_gappers
|
|
from strategy.evaluation.structure import opening_range
|
|
|
|
# don't arm new triggers on stale data when blocking is enabled — a break
|
|
# computed off a frozen feed could fire into a name that has already moved.
|
|
if _ingest_stale_block and _data_stale["v"]:
|
|
return "skipped: data stale (INGEST_STALE_BLOCK)"
|
|
session_date = _now_session_date()
|
|
# fresh catalysts (the discover step ingests the feeds; we just read them)
|
|
news_map = recent_news_map(
|
|
rt["research_con"], lookback_hours=int(os.environ.get("NEWS_LOOKBACK_HOURS", "8")))
|
|
news_boost = float(os.environ.get("NEWS_RANK_BOOST", "2.0"))
|
|
# GATED graded ranking from the LLM catalyst read. When the score blend is
|
|
# off, ranking is the existing binary `news_boost if catalyst else 1.0`.
|
|
# When on, a bullish catalyst boosts by `1 + gain*score` and a confirmed
|
|
# dilutive offering is sunk below the armed cut (multiplied by a small
|
|
# penalty) so the book stops arming the gap-and-fade traps.
|
|
cat_map: dict = {}
|
|
if ollama_cfg.catalyst_score_enabled:
|
|
from research.ingestion.news_enrichment import catalyst_map as _cmap
|
|
try:
|
|
cat_map = _cmap(rt["research_con"])
|
|
except Exception: # noqa: BLE001
|
|
cat_map = {}
|
|
news_rank_gain = float(os.environ.get("NEWS_RANK_BOOST_GAIN", "2.0"))
|
|
dilutive_penalty = float(os.environ.get("NEWS_DILUTIVE_PENALTY", "0.1"))
|
|
|
|
def _cat_mult(sym: str, has_headline: bool) -> float:
|
|
if not ollama_cfg.catalyst_score_enabled:
|
|
return news_boost if has_headline else 1.0
|
|
from research.ingestion.news_enrichment import catalyst_score as _cscore
|
|
adv = cat_map.get((sym or "").upper())
|
|
if not adv:
|
|
return 1.0
|
|
if adv.get("is_dilutive") and float(adv.get("conviction") or 0.0) >= (
|
|
ollama_cfg.dilution_veto_min_conviction):
|
|
return dilutive_penalty
|
|
return 1.0 + news_rank_gain * (_cscore(adv) or 0.0)
|
|
# optional MANUAL per-symbol priority nudge (SYMBOL_WEIGHTS=ABCD:1.2,XYZ:1.1)
|
|
# — a small boost to names you want watched; keep modest so it doesn't
|
|
# override the gap/volume logic.
|
|
weights: dict[str, float] = {}
|
|
for part in os.environ.get("SYMBOL_WEIGHTS", "").split(","):
|
|
if ":" in part:
|
|
s, w = part.split(":", 1)
|
|
try:
|
|
weights[s.strip().upper()] = float(w)
|
|
except ValueError:
|
|
pass
|
|
candidates: list[dict] = []
|
|
# rank ALL session symbols by gap*rvol (thresholds applied later, per
|
|
# trigger, so the board still shows the field pre-gap)
|
|
gappers = scan_gappers(
|
|
rt["research_con"], session_date,
|
|
min_gap_pct=0.0, min_relative_volume=0.0,
|
|
price_min=price_min_v, price_max=price_max_v,
|
|
limit=max(12, book.max_armed + 6),
|
|
)
|
|
for g in gappers:
|
|
bars = rq.query_minute_bars(rt["research_con"], g.symbol, session_date)
|
|
hi, lo, complete = opening_range(bars, orb_bars=orb_bars_v)
|
|
stop = lo * (1.0 - stop_cushion_v) if lo else None
|
|
catalyst = news_map.get(g.symbol, "")
|
|
# recent $-volume (last 5 regular-hours bars) — the liquidity gate.
|
|
# rvol is relative; this is the absolute "can it absorb an order" check.
|
|
dollar_vol = 0.0
|
|
day_open = None
|
|
try:
|
|
rth = bars[bars["is_regular_hours"]] if "is_regular_hours" in bars else bars
|
|
tail = rth.tail(5)
|
|
dollar_vol = float((tail["close"] * tail["volume"]).sum())
|
|
if len(rth): # session open = first regular-hours bar's open
|
|
day_open = float(rth.iloc[0]["open"])
|
|
except Exception: # noqa: BLE001
|
|
dollar_vol = 0.0
|
|
# STATIC overnight gap (session open vs prev close) — the population
|
|
# definition. The scanner's g.gap_pct is DYNAMIC (latest price vs prev
|
|
# close): intraday rallies masqueraded as gappers through it (TVRD 6%
|
|
# true gap entered 2026-07-07; audit 2026-07-08). None until the open.
|
|
gap_open_pct = None
|
|
if day_open and g.prev_close and g.prev_close > 0:
|
|
gap_open_pct = (day_open / g.prev_close - 1.0) * 100.0
|
|
candidates.append({
|
|
"symbol": g.symbol,
|
|
"gap_pct": g.gap_pct,
|
|
"gap_open_pct": gap_open_pct,
|
|
"rvol": g.relative_volume,
|
|
"trigger": hi,
|
|
"stop": stop,
|
|
"range_pct": ((hi - lo) / hi) if (hi and lo and hi > 0) else 0.0,
|
|
"cum_volume": g.cumulative_volume,
|
|
"dollar_vol": dollar_vol,
|
|
"catalyst": catalyst,
|
|
"day_open": day_open,
|
|
# a fresh catalyst is WHY a small-cap runs — boost it up the rank;
|
|
# times an optional manual per-symbol weight (default 1.0)
|
|
"_score": (g.gap_pct * max(g.relative_volume, 0.1)
|
|
* _cat_mult(g.symbol, bool(catalyst))
|
|
* weights.get(g.symbol, 1.0)),
|
|
"complete": complete,
|
|
})
|
|
# pre-open / before any gappers form: surface the queued watchlist so the
|
|
# board shows what we're about to watch instead of looking blank
|
|
if not candidates:
|
|
for s in symbols[: book.max_armed]:
|
|
candidates.append({"symbol": s, "gap_pct": 0.0, "rvol": 0.0,
|
|
"trigger": None, "stop": None,
|
|
"range_pct": 0.0, "complete": False})
|
|
# catalyst-backed gappers float to the top of the armed set
|
|
candidates.sort(key=lambda c: c.get("_score", 0.0), reverse=True)
|
|
book.arm(candidates)
|
|
armed = sum(1 for t in book.triggers.values() if t.state == "armed")
|
|
n_news = sum(1 for c in candidates if c.get("catalyst"))
|
|
return f"tracking={len(book.triggers)} armed={armed}" + (f" news={n_news}" if n_news else "")
|
|
|
|
def step_quotes():
|
|
"""Record NBBO bid/ask SIZES for tracked + held names (order-book-imbalance
|
|
proxy — full L2 depth needs a paid feed; NBBO sizes are the free approximation).
|
|
Recording only, zero trading impact; premarket included (that's where the
|
|
edge data lives). A future shadow study tests ask/bid-size imbalance as a
|
|
pop-vs-fade predictor."""
|
|
syms = set(book.triggers.keys())
|
|
if rt["execution"] is not None:
|
|
try:
|
|
syms |= set(rt["execution"]._held_symbols())
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
if not syms:
|
|
return None
|
|
try:
|
|
quotes = client.get_latest_quotes(sorted(syms))
|
|
except Exception as exc: # noqa: BLE001
|
|
return f"quotes fetch failed: {exc}"
|
|
n = 0
|
|
for sym, q in (quotes or {}).items():
|
|
if not q:
|
|
continue
|
|
rt["research_con"].execute(
|
|
"INSERT INTO quote_snapshots (symbol, bid_price, bid_size, ask_price, ask_size) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
[sym, q.get("bp"), q.get("bs"), q.get("ap"), q.get("as")])
|
|
n += 1
|
|
return f"{n} quotes ({len(syms)} names)"
|
|
|
|
# The trigger runs on its OWN thread, not the scheduler, so a slow ingest
|
|
# can never freeze breakout detection. It only touches the (thread-safe)
|
|
# book, execution service, and event store — never research_con.
|
|
trigger_interval = float(os.environ.get("TRIGGER_INTERVAL_SECONDS", "4"))
|
|
stop_event = threading.Event()
|
|
|
|
def _batch_prices(syms):
|
|
"""One batched last-trade call for all armed names (8 syms ≈ 1 sym cost)."""
|
|
if client is None or not syms:
|
|
return {}
|
|
try:
|
|
trades = client.get_latest_trades(list(syms))
|
|
except Exception: # noqa: BLE001
|
|
return {}
|
|
out = {}
|
|
for s in syms:
|
|
tr = trades.get(s) or trades.get(s.upper())
|
|
if tr and tr.get("p") is not None:
|
|
try:
|
|
out[s] = float(tr["p"])
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return out
|
|
|
|
_block_logged: set[str] = set() # dedup TRIGGER BLOCKED spam (re-checked every pass)
|
|
|
|
def _trigger_pass():
|
|
snap = book.snapshot()
|
|
syms = [r["symbol"] for r in snap]
|
|
if not syms:
|
|
return
|
|
for s, px in _batch_prices(syms).items():
|
|
book.update_price(s, px)
|
|
fired: list[str] = []
|
|
blocked: list[str] = []
|
|
if rt["execution"] is not None:
|
|
try:
|
|
book.mark_filled(rt["execution"]._held_symbols())
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
for t in book.fires():
|
|
res = rt["execution"].submit_breakout_now(
|
|
t.symbol, t.trigger, t.stop, last_price=t.price,
|
|
cum_volume=t.cum_volume, halted=_recently_halted(t.symbol),
|
|
day_open=t.day_open, rank=t.rank,
|
|
# STATIC overnight gap only — never the dynamic book gap
|
|
gap_pct=(t.gap_open_pct / 100.0) if t.gap_open_pct is not None else None,
|
|
rvol=t.rvol, price=t.price,
|
|
)
|
|
if res.get("ok"):
|
|
book.mark_fired(t.symbol)
|
|
fired.append(t.symbol)
|
|
elif res.get("skipped") in ("over_extended", "over_extended_day",
|
|
"halted_symbol"):
|
|
# surface the anti-chase / halt skips once per symbol (they
|
|
# re-check every 4s; routine dedup skips stay quiet entirely)
|
|
key = f"{t.symbol}:{res['skipped']}"
|
|
if key not in _block_logged:
|
|
_block_logged.add(key)
|
|
blocked.append(key)
|
|
snap = book.snapshot()
|
|
armed_n = sum(1 for r in snap if r["state"] == "armed")
|
|
rt["store"].emit(
|
|
ModuleTickEvent(
|
|
timestamp=datetime.now(), mode=EventMode.PAPER,
|
|
correlation_id=rt["session_id"],
|
|
message=f"triggers armed={armed_n} fired={len(fired)}",
|
|
module="triggers", stage="snapshot", duration_ms=0.0,
|
|
input_count=len(syms), output_count=len(fired),
|
|
metrics={"triggers": snap, "armed": armed_n}, errors=[],
|
|
)
|
|
)
|
|
if fired:
|
|
print(f"[{datetime.now():%H:%M:%S}] TRIGGER FIRED {', '.join(fired)}", flush=True)
|
|
if blocked:
|
|
print(f"[{datetime.now():%H:%M:%S}] TRIGGER BLOCKED {', '.join(blocked)}", flush=True)
|
|
|
|
def trigger_loop():
|
|
while not stop_event.wait(trigger_interval):
|
|
try:
|
|
_trigger_pass()
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[trigger] error: {exc}", flush=True)
|
|
|
|
# --- live exit management (breakeven / trail / scale / first-red) -------
|
|
exit_cfg = ExitConfig.from_env()
|
|
|
|
def _rth_bars(sym):
|
|
from research import query as rq
|
|
b = rq.query_minute_bars(rt["research_con"], sym, _now_session_date())
|
|
if b is not None and len(b) and "is_regular_hours" in b.columns:
|
|
b = b[b["is_regular_hours"] == True].reset_index(drop=True) # noqa: E712
|
|
return b
|
|
|
|
exit_mgr = (
|
|
LiveExitManager(client, rt["store"], _rth_bars, cfg=exit_cfg,
|
|
session_id=rt["session_id"])
|
|
if client else None
|
|
)
|
|
print(f"[boot] exit management: {exit_cfg.describe()}")
|
|
_es = rt["execution"].settings if rt["execution"] is not None else None
|
|
if _es is not None:
|
|
print(
|
|
f"[boot] entry gates: VWAP {'ENFORCED' if _es.require_above_vwap else 'shadow-only'}"
|
|
f", anti-chase {'ON (auto path)' if _es.unified_entry else 'fast-path only'}"
|
|
f", fill={_es.entry_fill_model}"
|
|
f", daily-cap {_es.max_fresh_entries_per_day or 'off'}"
|
|
f", risk-cap ${_es.max_risk_dollars:.0f}"
|
|
)
|
|
if _es.concentrate_top_n > 0:
|
|
print(
|
|
"[boot] WARNING: TRADING_CONCENTRATE_TOP_N>0, but rank-concentration is "
|
|
"enforced ONLY on the dead fast-trigger path, NOT the live auto-approval "
|
|
"path — it will not affect live entries (see entry-path-fast-vs-auto)."
|
|
)
|
|
|
|
# Telegram push alerts — gated, decoupled (polls the event store), and
|
|
# fire-and-forget so it can never impede trading.
|
|
from runtime.notifier import TelegramNotifier, send_telegram
|
|
notifier = TelegramNotifier(rt["store"])
|
|
if notifier.enabled:
|
|
print("[boot] telegram alerts: ON")
|
|
send_telegram("\U0001F916 Momentum loop started — push alerts armed.")
|
|
else:
|
|
print("[boot] telegram alerts: off (TELEGRAM_BOT_TOKEN/CHAT_ID unset)")
|
|
|
|
def step_notify():
|
|
n = notifier.poll()
|
|
return f"telegram: {n} alert(s)" if n else None
|
|
|
|
def step_manage_exits():
|
|
if exit_mgr is None:
|
|
return None
|
|
acts = exit_mgr.manage()
|
|
return ", ".join(acts) if acts else None
|
|
|
|
# --- end-of-day auto-flatten (a day-trading book shouldn't gap overnight) --
|
|
eod_enabled = _flag("TRADING_EOD_FLATTEN_ENABLED", "1")
|
|
# Legacy absolute time (15:55) is kept ONLY to derive the normal lead minutes
|
|
# (16:00 - 15:55 = 5) — the window itself is now relative to the ACTUAL close
|
|
# so it fires on early-close half-days too. Before a multi-day closure the
|
|
# lead widens so the book is flat with margin (the ATPC naked-holiday case).
|
|
try:
|
|
eod_h, eod_m = (int(x) for x in
|
|
os.environ.get("TRADING_EOD_FLATTEN_TIME", "15:55").split(":"))
|
|
except Exception: # noqa: BLE001
|
|
eod_h, eod_m = 15, 55
|
|
_normal_lead = max(1, 16 * 60 - (eod_h * 60 + eod_m)) # 15:55 -> 5
|
|
_pre_lead = int(os.environ.get("TRADING_PRE_CLOSURE_FLATTEN_LEAD_MIN", "20"))
|
|
_eod_done = {"v": False}
|
|
|
|
def step_eod_flatten():
|
|
if not eod_enabled or _eod_done["v"] or rt["execution"] is None:
|
|
return None
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
now_et = datetime.now(ZoneInfo("America/New_York"))
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
# Fire inside [close - lead, close): after the close a market order just
|
|
# queues uselessly. The lead WIDENS before a multi-day closure so a thin
|
|
# name has time to fill and a transient failure still has a wide retry
|
|
# window. Skips weekends/holidays (session_close_hm -> None) — flattening
|
|
# into a closed market can't fill the sell.
|
|
due, pre_closure = eod_flatten_status(now_et, _normal_lead, _pre_lead)
|
|
if not due:
|
|
return None
|
|
reason = "eod_flatten_preclose" if pre_closure else "eod_flatten"
|
|
res = rt["execution"].close_session(reason)
|
|
# only mark done if it actually succeeded — otherwise RETRY on the next
|
|
# pass within the window (a flatten that errored once left positions naked
|
|
# overnight). close_session re-blocks entries each call, so retrying is safe.
|
|
ok = not res["errors"]
|
|
_eod_done["v"] = ok
|
|
closed = ", ".join(res["closed_positions"]) or "none"
|
|
tag = "PRE-HOLIDAY EOD FLATTEN" if pre_closure else "EOD FLATTEN"
|
|
if res["errors"]:
|
|
# A failed flatten before a MULTI-DAY gap = naked carry across the
|
|
# closure (the silent 6/18 Juneteenth failure). Scream about it.
|
|
gap = days_to_next_session(now_et.date())
|
|
sev = f"‼ NAKED-CARRY RISK ({gap}-day gap ahead): " if pre_closure else ""
|
|
print(f"[{datetime.now():%H:%M:%S}] {sev}{tag} could not flatten: {res['errors']}", flush=True)
|
|
send_telegram(f"⚠️ {tag} FAILED to flatten: {res['errors']} — {sev or 'will retry'}")
|
|
return f"{sev}{tag} closed [{closed}] errors={res['errors']} (will retry)"
|
|
# success — push the once-per-day EOD P&L summary
|
|
try:
|
|
from storage.projections import query_session_pnl
|
|
notifier.send_eod_summary(now_et.date().isoformat(),
|
|
query_session_pnl(rt["store"]))
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return f"{tag} closed [{closed}]"
|
|
|
|
# --- open catch-up flatten: a day-trading book should be FLAT by the open;
|
|
# close anything carried from a prior session (a missed/failed EOD flatten)
|
|
# before it bleeds for days — the ATPC-over-Juneteenth backstop. Does NOT
|
|
# halt the session; today's fresh names are left to trade normally.
|
|
catchup_enabled = _flag("TRADING_OPEN_CATCHUP_FLATTEN_ENABLED", "1")
|
|
_catchup_done = {"v": False}
|
|
|
|
def step_open_catchup_flatten():
|
|
if not catchup_enabled or _catchup_done["v"] or rt["execution"] is None:
|
|
return None
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
now_et = datetime.now(ZoneInfo("America/New_York"))
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
if not is_regular_hours(now_et): # only act while the market is open
|
|
return None
|
|
res = rt["execution"].flatten_overnight_carries(now_et.date(), "open_catchup")
|
|
if res["errors"]:
|
|
print(f"[{datetime.now():%H:%M:%S}] ‼ OPEN CATCH-UP could not flatten carry: "
|
|
f"{res['errors']}", flush=True)
|
|
return f"OPEN CATCH-UP carries={res['carries']} closed={res['closed']} errors={res['errors']} (will retry)"
|
|
# clean (no carries) or all carries closed -> done for the session
|
|
_catchup_done["v"] = True
|
|
if res["carries"]:
|
|
print(f"[{datetime.now():%H:%M:%S}] OPEN CATCH-UP flattened stale carries: "
|
|
f"{res['closed']}", flush=True)
|
|
return f"OPEN CATCH-UP flattened overnight carries {res['closed']}"
|
|
return None
|
|
|
|
def step_news():
|
|
"""Land latest Alpaca/Benzinga news for the tracked universe into
|
|
raw_news_items (feeds recent_news_map catalyst prioritisation + the Ollama
|
|
enrichment). Off the hot path; never blocks trading."""
|
|
if not client:
|
|
return "skipped (no keys)"
|
|
res = ingest_alpaca_news(
|
|
rt["research_con"], client, symbols=symbols,
|
|
limit=int(os.environ.get("ALPACA_NEWS_LIMIT", "50")),
|
|
)
|
|
if res.error:
|
|
return f"ERROR {res.error}"
|
|
if res.new_items:
|
|
return f"+{len(res.new_items)} new / {res.item_count} fetched"
|
|
return None
|
|
|
|
# Local-LLM (Ollama) news/catalyst enrichment — advisory layer. Runs on its
|
|
# OWN daemon thread with its OWN DB connection (started below), NOT the
|
|
# scheduler thread: a slow/hung Ollama pass (up to batch_limit x timeout) must
|
|
# never stall exits / guard / eod-flatten, which share the scheduler thread,
|
|
# and research_con is a single psycopg2 connection unsafe to share across
|
|
# threads. Gated by ollama.enabled AND enrichment_enabled; no-op if Ollama down.
|
|
def enrich_loop():
|
|
try:
|
|
econ = open_research_db("market")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[enrich] no DB connection, enrichment off: {exc}", flush=True)
|
|
return
|
|
while not stop_event.wait(float(ollama_cfg.enrichment_interval_seconds)):
|
|
try:
|
|
enrich_recent_news(econ, ollama_cfg)
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[enrich] error: {exc}", flush=True)
|
|
|
|
scheduler = Scheduler()
|
|
scheduler.add("news", step_news, float(os.environ.get("ALPACA_NEWS_INTERVAL_SECONDS", "120")),
|
|
enabled=_flag("ALPACA_NEWS_ENABLED", "1"))
|
|
scheduler.add("discover", step_discover,
|
|
float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")),
|
|
enabled=_flag("DISCOVER_ENABLED", "1"))
|
|
scheduler.add("vratio", step_vratio, float(os.environ.get("VOLUME_RATIO_INTERVAL_SECONDS", "300")),
|
|
enabled=_flag("VOLUME_RATIO_ENABLED", "1"))
|
|
scheduler.add("ingest", step_ingest, float(os.environ.get("LIVE_BARS_INTERVAL_SECONDS", "60")),
|
|
enabled=_flag("LIVE_BARS_ENABLED", "1"))
|
|
scheduler.add("watch", step_watch, float(os.environ.get("WATCHER_INTERVAL_SECONDS", "30")),
|
|
enabled=_flag("WATCHER_ENABLED", "1"))
|
|
scheduler.add("arm", step_arm, float(os.environ.get("TRIGGER_ARM_INTERVAL_SECONDS", "20")),
|
|
enabled=_flag("TRIGGER_ENABLED", "1"))
|
|
scheduler.add("sync", step_sync, float(os.environ.get("ALPACA_PAPER_SYNC_INTERVAL_SECONDS", "60")),
|
|
enabled=_flag("ALPACA_PAPER_SYNC_ENABLED", "1"))
|
|
scheduler.add("quotes", step_quotes, float(os.environ.get("QUOTE_SNAPSHOT_INTERVAL_SECONDS", "15")),
|
|
enabled=_flag("QUOTE_SNAPSHOT_ENABLED", "1"))
|
|
scheduler.add("execute", step_execute, float(os.environ.get("TRADING_EXECUTION_INTERVAL_SECONDS", "30")),
|
|
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
|
scheduler.add("guard", step_guard, float(os.environ.get("TRADING_ENTRY_GUARD_INTERVAL_SECONDS", "5")),
|
|
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
|
scheduler.add("exits", step_manage_exits, float(os.environ.get("TRADING_EXIT_MANAGE_INTERVAL_SECONDS", "12")),
|
|
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
|
scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")),
|
|
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
|
scheduler.add("notify", step_notify, float(os.environ.get("TELEGRAM_NOTIFY_INTERVAL_SECONDS", "45")),
|
|
enabled=notifier.enabled)
|
|
scheduler.add("catchup", step_open_catchup_flatten,
|
|
float(os.environ.get("TRADING_OPEN_CATCHUP_INTERVAL_SECONDS", "30")),
|
|
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
|
|
|
def run_pass():
|
|
# force every task due, in pipeline order
|
|
for task in scheduler.tasks:
|
|
task.last_run = -10**9
|
|
results = {}
|
|
for name in ("discover", "ingest", "watch", "arm", "sync", "execute", "exits", "eod", "catchup"):
|
|
for task in scheduler.tasks:
|
|
if task.name == name and task.enabled:
|
|
results.update(scheduler_run_one(task))
|
|
stamp = datetime.now().strftime("%H:%M:%S")
|
|
line = " | ".join(f"{k}: {v}" for k, v in results.items())
|
|
print(f"[{stamp}] {line}")
|
|
|
|
def scheduler_run_one(task):
|
|
task.last_run = time.monotonic()
|
|
task.run_count += 1
|
|
try:
|
|
out = task.func()
|
|
task.last_error = None
|
|
return {task.name: out}
|
|
except Exception as exc: # noqa: BLE001
|
|
task.error_count += 1
|
|
task.last_error = str(exc)
|
|
return {task.name: f"ERROR {exc}"}
|
|
|
|
if args.once:
|
|
run_pass()
|
|
if server:
|
|
server.shutdown()
|
|
return 0
|
|
|
|
print("[boot] entering live loop (Ctrl-C to stop)")
|
|
# start the fast trigger on its own thread so slow ingest can't freeze it
|
|
trigger_thread = None
|
|
if _flag("TRIGGER_ENABLED", "1") and client is not None:
|
|
trigger_thread = threading.Thread(target=trigger_loop, name="trigger", daemon=True)
|
|
trigger_thread.start()
|
|
print(f"[boot] fast trigger thread started (every {trigger_interval:.0f}s)")
|
|
# news enrichment on its OWN thread (never blocks the scheduler's trade steps)
|
|
if ollama_cfg.enabled and ollama_cfg.enrichment_enabled:
|
|
threading.Thread(target=enrich_loop, name="enrich", daemon=True).start()
|
|
print(f"[boot] news enrichment thread started (every {ollama_cfg.enrichment_interval_seconds}s, model {ollama_cfg.model})")
|
|
try:
|
|
# one immediate full pass, then interval-driven
|
|
run_pass()
|
|
while True:
|
|
results = scheduler.run_pending()
|
|
if results:
|
|
stamp = datetime.now().strftime("%H:%M:%S")
|
|
line = " | ".join(
|
|
f"{k}: {v if not isinstance(v, Exception) else f'ERROR {v}'}"
|
|
for k, v in results.items()
|
|
)
|
|
print(f"[{stamp}] {line}")
|
|
time.sleep(1.0)
|
|
except KeyboardInterrupt:
|
|
print("\n[stop] shutting down")
|
|
finally:
|
|
stop_event.set()
|
|
if trigger_thread is not None:
|
|
trigger_thread.join(timeout=5)
|
|
if server:
|
|
server.shutdown()
|
|
rt["store"].close()
|
|
rt["research_con"].close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|