momentum devandClaude Opus 4.8
3b692586f2
The owner bought Alpaca Algo Trader Plus (real-time SIP; entitlement propagating on the same keys — latest-trade already SIP-live, quotes/bars still 403). Before flipping ALPACA_DATA_FEED=sip, two guards so the free-tier ratio estimator can never double-count: - ingest_live_minute_bars resolves the EFFECTIVE feed and refuses volume_scale for anything non-IEX (provider marked alpaca_<feed>). - step_vratio self-disables when client.settings.feed != iex (saves the API calls too). The estimator stays as the dormant free-tier fallback (plan lapse -> flip env back). Test: SIP client + x25 ratios -> volumes NOT re-scaled, provider alpaca_sip. 4 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1176 lines
54 KiB
Python
1176 lines
54 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 IEX bars into research market.duckdb
|
|
(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")),
|
|
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
|
|
candidates.append({
|
|
"symbol": g.symbol,
|
|
"gap_pct": g.gap_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,
|
|
)
|
|
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())
|