""" top_stocks.py Runs whenever a position closes and a slot opens up. Re-scans CURRENT market conditions (not the stale 09:25 list) to find the strongest actionable candidate right now, respecting symbol cooldowns and currently-open positions. [FEATURE 2026-08-17] The candidate pool searched here is now premarket_20 (the full morning list of 20), not final_10 -- per the project's updated intraday-scanner requirement, premarket_20 is the primary intraday universe. Ranking now uses intraday_health.py's continuous health score (via evaluate_symbol()) rather than a one-shot score_premarket_candidate() call, so the same continuously-updated health state that decides entry eligibility also decides replacement ranking -- one source of truth, not two competing scores. Symbols health.evaluate_symbol() confirms as STALE/UNHEALTHY/REMOVED are excluded from replacement selection entirely. [FEATURE 2026-08-18] Every symbol in premarket_20 -- not just the ones currently eligible to fill a slot -- gets its health recomputed and persisted every time find_replacement() runs. Previously, a symbol sitting in exclude_symbols (currently open) or on cooldown never got recomputed during a replacement search, so its state/intraday_health.json entry could go stale until the next periodic 60-second cycle even though other parts of the system (the entry shortlist ranking, the dashboard) read that same file. Now every slot-open event doubles as a full health refresh for the whole pool, and selection among the symbols that CAN fill the slot is still: highest health_score wins, total_score only breaks a tie. Writes the running result to state/top_stocks.json (write_top_stocks) so the whole system has a single canonical file of "what looks best right now," as required by the project spec. The returned candidate still must pass fast_pipeline.evaluate_fast_pipeline() before monitor.py actually buys it — this module only ranks, it never enters directly. """ from datetime import datetime, timezone, timedelta from config_loader import get_config from logger_setup import get_logger from alpaca_client import get_client from scorer import score_premarket_candidate # still used for the informational total_score import symbol_filters import intraday_health import data_store log = get_logger("top_stocks") def _bars_to_dicts(bars) -> list: out = [] for b in bars: out.append({ "t": b.timestamp, "o": float(b.open), "h": float(b.high), "l": float(b.low), "c": float(b.close), "v": float(b.volume), }) return out def _is_on_cooldown(symbol: str, cooldowns: dict) -> bool: ts = cooldowns.get(symbol) if not ts: return False cooldown_minutes = get_config()["risk"]["symbol_cooldown_minutes"] expires = datetime.fromisoformat(ts) + timedelta(minutes=cooldown_minutes) return datetime.now(timezone.utc) < expires def find_replacement(exclude_symbols: list = None, candidate_pool: list = None, volume_baselines: dict = None, volatility_baselines: dict = None) -> dict: """ exclude_symbols: currently-open symbols, always excluded from filling the slot (but still get a fresh health refresh below -- see [FEATURE 2026-08-18]). candidate_pool: optional restricted pool; if None, defaults to state/watchlist.json's premarket_20 (falling back to final_10 only if premarket_20 is somehow empty -- e.g. a very early call before the 09:00 scan has ever written the file). [FEATURE 2026-08-18] Every symbol in the pool -- including ones currently open or on cooldown, not just the ones that could fill THIS slot -- gets a fresh intraday_health.evaluate_symbol() call and a persisted state/intraday_health.json update before any selection happens. Previously, a symbol sitting in exclude_symbols/cooldown never got its health recomputed during a replacement search (its entry in the state file could go stale until the next periodic 60-second eval cycle), even though that same file also feeds _scan_for_entries()' shortlist ranking and the dashboard. Refreshing the whole pool here means every slot-open event doubles as a full health refresh for premarket_20, not just for the symbols eligible to fill it right now. Selection itself is unchanged: among symbols that ARE eligible to fill the slot (not excluded, not on cooldown, health-eligible per that just-refreshed reading), the highest health_score wins, total_score only breaks a tie. """ cfg = get_config() exclude_symbols = set(exclude_symbols or []) cooldowns = data_store.load_cooldowns() health_state = data_store.load_health_state() if candidate_pool is None: watchlist = data_store.load_watchlist() candidate_pool = watchlist.get("premarket_20", []) or watchlist.get("final_10", []) candidate_pool = [c["symbol"] if isinstance(c, dict) else c for c in candidate_pool] if not candidate_pool: log.info("[REPLACEMENT] No candidate pool available to search") return None client = get_client() now = datetime.now(timezone.utc) start = now - timedelta(hours=cfg["intraday_health"]["full_rescan_lookback_hours"]) # [Phase 5] Real per-symbol historical volume when the caller has it # (monitor.py's cached premarket-scan baselines), falling back to the # old flat universe constant per-symbol only where missing -- same # fix as premarket_scanner.scan()'s volume_baselines param. fallback_baseline = cfg["universe"]["min_avg_daily_volume"] volume_baselines = volume_baselines or {} volatility_baselines = volatility_baselines or {} log.info(f"[REPLACEMENT] Refreshing health for all {len(candidate_pool)} " f"premarket_20 symbols before selecting a fill for the open slot") # ---- Pass 1: refresh health for the ENTIRE pool, unconditionally ---- health_updates = {} fresh = {} # symbol -> (HealthReading, bars) for symbol in candidate_pool: # Defensive re-check: even though candidate_pool should already # be pre-filtered by premarket_scanner.py, re-verify here so a # config change mid-day (or a future caller passing an # unfiltered pool) can never let an ETF/leveraged/complex-name # symbol slip into a same-day replacement. asset = client.get_asset(symbol) asset_name = getattr(asset, "name", "") if asset else "" ok, reason = symbol_filters.passes_symbol_filters(symbol, asset_name) if not ok: log.debug(f"[REPLACEMENT] {symbol} skipped: {reason}") continue bars = _bars_to_dicts(client.get_minute_bars(symbol, start, now)) if len(bars) < 3: log.debug(f"[REPLACEMENT] {symbol} skipped: insufficient bars for a fresh health check") continue reading, new_health_entry = intraday_health.evaluate_symbol( symbol, bars, volume_baselines.get(symbol, fallback_baseline), health_state) health_updates[symbol] = new_health_entry fresh[symbol] = (reading, bars) if health_updates: health_state.update(health_updates) data_store.save_health_state(health_state) log.info(f"[REPLACEMENT] Refreshed health for {len(health_updates)}/" f"{len(candidate_pool)} pool symbols") # ---- Pass 2: among the just-refreshed pool, who can actually fill # this slot? (not open, not on cooldown, health-eligible) ---- scored = [] for symbol, (reading, bars) in fresh.items(): if symbol in exclude_symbols: continue if _is_on_cooldown(symbol, cooldowns): log.debug(f"[REPLACEMENT] {symbol} skipped: on cooldown") continue if not intraday_health.is_eligible_for_entry(reading.confirmed_state): log.debug(f"[REPLACEMENT] {symbol} skipped: health={reading.confirmed_state}") continue session_high = max(b["h"] for b in bars) quote = client.get_latest_quote(symbol) bid = float(quote.bid_price) if quote and quote.bid_price else None ask = float(quote.ask_price) if quote and quote.ask_price else None result = score_premarket_candidate(symbol, bars, session_high, min(b["l"] for b in bars), volume_baselines.get(symbol, fallback_baseline), bid, ask, daily_atr_pct=volatility_baselines.get(symbol)) # [FEATURE 2026-09-15] Same structural volatility floor scan() # enforces -- candidate_pool is normally already prefiltered by # premarket_scanner.scan(), but this is a cheap defense-in-depth # check for a caller that passes its own unfiltered pool (see # this function's docstring re: symbol_filters re-check above). if not result["flags"].get("meets_min_volatility", True): log.debug(f"[REPLACEMENT] {symbol} skipped: below min daily ATR% " f"(structurally low volatility, e.g. BDC/SPAC)") continue result["pm_high"] = session_high # reused as intraday resistance reference result["health_score"] = reading.health_score result["health_state"] = reading.confirmed_state result["price_slope"] = reading.price_slope_class result["vwap_slope"] = reading.vwap_slope_class scored.append(result) if not scored: log.info("[REPLACEMENT] No actionable (health-eligible) candidates found") return None # Rank primarily by the just-refreshed continuous health score -- # the single source of truth for "which of the original candidates # is still the best one to fill an open slot" -- falling back to the # premarket-style total_score only to break a health-score tie. scored.sort(key=lambda r: (r["health_score"], r["total_score"]), reverse=True) best = scored[0] # [FEATURE 2026-08-25] Absolute floor on top of the relative ranking # above. 2026-08-24 session review flagged RKT filling a slot at # health_score=27.57 -- dramatically below anything else this # project has let through any gate -- because ranking alone only # ever asks "who's the best OF THE CANDIDATES HERE," never "is even # the best one actually good enough." A weak field can still # produce a genuinely bad pick. When nothing clears the floor, # leave the slot empty this cycle rather than filling it with the # least-bad weak option -- find_replacement() runs again the next # time a slot frees up or the pool refreshes, so an empty slot now # isn't a permanent loss, just a pass on today's weak field. min_replacement_health_score = cfg.get("trading", {}).get("min_replacement_health_score", 0) if best["health_score"] < min_replacement_health_score: log.info(f"[REPLACEMENT] Best candidate {best['symbol']} (health_score=" f"{best['health_score']}) is below min_replacement_health_score=" f"{min_replacement_health_score} -- leaving slot empty this cycle " f"rather than filling it with a weak pick") data_store.write_top_stocks(scored) return None log.info(f"[REPLACEMENT] Running top_stocks.py") log.info(f"[REPLACEMENT] {best['symbol']} selected (health={best['health_state']}, " f"health_score={best['health_score']}, score={best['total_score']})") data_store.write_top_stocks(scored) return best def register_cooldown(symbol: str): cooldowns = data_store.load_cooldowns() cooldowns[symbol] = datetime.now(timezone.utc).isoformat() data_store.save_cooldowns(cooldowns) if __name__ == "__main__": result = find_replacement() if result: print(f"Best replacement: {result['symbol']} score={result['total_score']}") else: print("No replacement candidate found.")