main.py

import os
import json
from dotenv import load_dotenv
load_dotenv(dotenv_path=os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
import logging
from utils import get_sgt_now
import random
import sqlite3
import time
from datetime import datetime, timedelta

import numpy as np
import pandas as pd
import yaml
from ingest_prices import PriceIngestor
from ingest_news import NewsIngestor
from positions import PositionTracker
from portfolio import PortfolioManager
from state_recovery import recover_positions_from_seed, enforce_position_cap, purge_seeded_open_positions
from broker import AlpacaBroker, AlpacaError, create_broker_from_env

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


def _get_config_path(config_path=None):
    if config_path:
        return config_path
    base_dir = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(base_dir, 'config.yaml')


def _load_config(config_path):
    with open(config_path, 'r') as f:
        payload = yaml.safe_load(f) or {}
    if isinstance(payload, dict):
        payload.setdefault("_config_path", os.path.abspath(config_path))
        payload.setdefault("_config_base_dir", os.path.dirname(os.path.abspath(config_path)))
    return payload


def _resolve_path(base_dir, path_value):
    if os.path.isabs(path_value):
        return path_value
    return os.path.join(base_dir, path_value)



def _get_open_position_symbols(config_path, table_names=None):
    """Return distinct OPEN symbols across the requested position tables."""
    config = _load_config(config_path)
    
    if table_names is None:
        ai_cfg = config.get("ai_trading", {}) if isinstance(config, dict) else {}
        ai_disabled_by_env = str(os.getenv("DISABLE_AI_TRADING", "")).strip().lower() in {"1", "true", "yes", "on"}
        ai_enabled = bool(ai_cfg.get("enabled", False)) and not ai_disabled_by_env
        
        core_disabled_by_env = str(os.getenv("DISABLE_CORE_TRADING", "")).strip().lower() in {"1", "true", "yes", "on"}
        
        table_names = []
        if not core_disabled_by_env:
            table_names.append("positions")
        if ai_enabled:
            table_names.append("positions_ai")
        if not table_names:
            table_names = ["positions"]

    base_dir = os.path.dirname(os.path.abspath(config_path))
    db_path = _resolve_path(base_dir, config["data"]["cache_path"])
    ordered = []
    seen = set()
    conn = sqlite3.connect(db_path)
    try:
        for table_name in table_names:
            safe_name = str(table_name or "").strip()
            if safe_name not in {"positions", "positions_ai"}:
                continue
            try:
                rows = conn.execute(
                    f"SELECT DISTINCT symbol FROM {safe_name} WHERE status='OPEN'"
                ).fetchall()
            except sqlite3.OperationalError:
                rows = []
            for row in rows:
                symbol = str((row[0] if row else "") or "").strip().upper()
                if not symbol or symbol in seen:
                    continue
                seen.add(symbol)
                ordered.append(symbol)
    finally:
        conn.close()
    return ordered



def _position_cash_metrics(positions_df, current_capital):
    """Return gross exposure and cash available for open positions.

    Long positions consume cash at their current mark (or entry mark when no
    current mark is available). Short positions are included in gross notional
    exposure but do not reduce cash in this simple daily-report view.
    """
    try:
        capital = float(current_capital or 0.0)
    except (TypeError, ValueError):
        capital = 0.0

    if positions_df is None or not hasattr(positions_df, "empty") or positions_df.empty:
        return 0.0, max(0.0, capital)

    gross_notional = 0.0
    long_notional = 0.0
    for _, row in positions_df.iterrows():
        try:
            quantity = abs(float(row.get("quantity", 0.0) or 0.0))
            mark = row.get("current_price")
            if mark is None or str(mark) == "nan":
                mark = row.get("entry_price")
            mark = float(mark or 0.0)
        except (TypeError, ValueError):
            continue
        if quantity <= 0.0 or mark <= 0.0:
            continue
        notional = quantity * mark
        gross_notional += notional
        side = str(row.get("side", "LONG") or "LONG").strip().upper()
        if side != "SHORT":
            long_notional += notional

    return gross_notional, max(0.0, capital - long_notional)

def _record_pipeline_issue(pipeline_stats, severity, source, message, max_items=20):
    """Record pipeline warnings/errors in a bounded structure that can be emailed."""
    if not isinstance(pipeline_stats, dict):
        return
    sev = str(severity or "ERROR").strip().upper()
    if sev not in {"ERROR", "WARNING"}:
        sev = "ERROR"

    key = "error_count" if sev == "ERROR" else "warning_count"
    pipeline_stats[key] = int(pipeline_stats.get(key, 0) or 0) + 1

    issues = pipeline_stats.get("issues")
    if not isinstance(issues, list):
        issues = []
        pipeline_stats["issues"] = issues

    issue = {
        "time": get_sgt_now().strftime("%H:%M:%S"),
        "severity": sev,
        "source": str(source or "pipeline"),
        "message": str(message or ""),
    }
    if len(issues) < int(max_items):
        issues.append(issue)
    else:
        pipeline_stats["issue_overflow_count"] = int(pipeline_stats.get("issue_overflow_count", 0) or 0) + 1


def _finalize_pipeline_health(pipeline_stats):
    if not isinstance(pipeline_stats, dict):
        return
    errors = int(pipeline_stats.get("error_count", 0) or 0)
    warnings = int(pipeline_stats.get("warning_count", 0) or 0)
    failed = int(pipeline_stats.get("tickers_failed", 0) or 0)
    if failed > errors:
        errors = failed
    pipeline_stats["error_count"] = errors
    pipeline_stats["warning_count"] = warnings
    pipeline_stats["run_health"] = "ERROR" if errors > 0 else ("WARNING" if warnings > 0 else "OK")


class DailyBacktester:
    def __init__(self, config_path=None):
        self.config_path = _get_config_path(config_path)
        self.config = _load_config(self.config_path)
        self.base_dir = os.path.dirname(os.path.abspath(self.config_path))
        self.db_path = _resolve_path(self.base_dir, self.config['data']['cache_path'])
        self.feature_store_dir = os.path.join(self.base_dir, 'feature_store')
        self.registry_path = os.path.join(self.base_dir, 'model_registry.json')
        self.models_dir = os.path.join(self.base_dir, 'models')
        self.results_dir = _resolve_path(
            self.base_dir,
            self.config.get('output', {}).get('results_dir', './results')
        )
        self.universe_path = _resolve_path(self.base_dir, self.config['universe']['source'])
        self.core_tracker = PositionTracker(self.config_path, table_name="positions")
        self.ai_tracker = PositionTracker(self.config_path, table_name="positions_ai")
        
        # Initialize broker if Alpaca mode is enabled
        self.broker_mode = os.environ.get("ARM_LIVE_TRADING", "simulation").lower()
        self.core_broker = None
        self.ai_broker = None
        if self.broker_mode == "alpaca":
            disable_core = str(os.environ.get("DISABLE_CORE_TRADING", "0")).strip() in {"1", "true", "yes"}
            disable_ai = str(os.environ.get("DISABLE_AI_TRADING", "0")).strip() in {"1", "true", "yes"}

            if not disable_core:
                try:
                    self.core_broker = create_broker_from_env("core")
                    account = self.core_broker.validate_connection()
                    logger.info(
                        "Alpaca Core connected: equity=$%.2f, cash=$%.2f, buying_power=$%.2f",
                        account["equity"],
                        account["cash"],
                        account["buying_power"],
                    )
                except Exception as e:
                    logger.error("Failed to connect Alpaca Core broker: %s", e)
                    raise RuntimeError(f"Alpaca Core broker connection failed in alpaca live trading mode: {e}")

            if not disable_ai:
                try:
                    self.ai_broker = create_broker_from_env("ai")
                    account = self.ai_broker.validate_connection()
                    logger.info(
                        "Alpaca AI connected: equity=$%.2f, cash=$%.2f, buying_power=$%.2f",
                        account["equity"],
                        account["cash"],
                        account["buying_power"],
                    )
                except Exception as e:
                    logger.error("Failed to connect Alpaca AI broker: %s", e)
                    raise RuntimeError(f"Alpaca AI broker connection failed in alpaca live trading mode: {e}")
        
        self.init_all_tables()

    def init_all_tables(self):
        """Ensure all required database tables exist before any processing starts."""
        from ingest_prices import PriceIngestor
        from ingest_news import NewsIngestor
        from positions import PositionTracker

        # Initialize core tables
        PriceIngestor(self.config_path).init_db()
        NewsIngestor(self.config_path).init_db()
        PositionTracker(self.config_path, table_name="positions")
        PositionTracker(self.config_path, table_name="positions_ai")
        logger.info("All database tables initialized successfully.")

    def _sync_sqlite_positions_with_alpaca(self, tracker, broker, trade_date_str):
        """Synchronize the local SQLite positions table with the broker's actual open positions."""
        if not broker:
            return
        try:
            logger.info("Synchronizing SQLite table %s with Alpaca actual positions...", tracker.table_name)
            alpaca_positions = broker.get_positions()
            
            # Create mapping of symbol -> position dict from Alpaca
            alpaca_map = {}
            for p in alpaca_positions:
                sym = str(p.get('symbol') or '').strip().upper()
                if sym:
                    alpaca_map[sym] = p

            # Fetch open positions from SQLite
            with sqlite3.connect(tracker.db_path) as conn:
                cursor = conn.cursor()
                sqlite_open = cursor.execute(
                    f"SELECT id, symbol, entry_price, quantity, side FROM {tracker.table_name} WHERE status='OPEN'"
                ).fetchall()
                
                sqlite_map = {row[1]: {'id': row[0], 'entry_price': row[2], 'quantity': row[3], 'side': row[4]} for row in sqlite_open}

                # 1. Close positions in SQLite that do not exist on Alpaca
                for sym, sql_pos in sqlite_map.items():
                    if sym not in alpaca_map:
                        logger.warning("Position %s is OPEN in SQLite but does not exist on Alpaca. Closing in SQLite at entry price.", sym)
                        # Close in SQLite at entry price to avoid phantom P&L
                        cursor.execute(
                            f"UPDATE {tracker.table_name} SET status='CLOSED', exit_date=?, exit_price=?, realized_pnl=0.0, exit_reason='alpaca_sync_close' WHERE id=?",
                            (trade_date_str, sql_pos['entry_price'], sql_pos['id'])
                        )

                # 2. Add or update positions in SQLite to match Alpaca
                for sym, alp_pos in alpaca_map.items():
                    alp_qty = float(int(abs(float(alp_pos.get('qty', 0.0)))))
                    alp_side = str(alp_pos.get('side') or 'LONG').upper()
                    alp_entry_price = float(alp_pos.get('avg_entry_price', 0.0) or alp_pos.get('price', 0.0))
                    if alp_qty <= 0.0:
                        continue

                    if sym in sqlite_map:
                        sql_pos = sqlite_map[sym]
                        # If quantity, side, or entry price differs significantly, update SQLite to match Alpaca
                        if abs(sql_pos['quantity'] - alp_qty) > 1e-5 or sql_pos['side'] != alp_side or abs(sql_pos['entry_price'] - alp_entry_price) > 0.01:
                            logger.info("Syncing existing position %s in SQLite: qty=%s -> %s, side=%s -> %s, entry=%s -> %s",
                                        sym, sql_pos['quantity'], alp_qty, sql_pos['side'], alp_side, sql_pos['entry_price'], alp_entry_price)
                            cursor.execute(
                                f"UPDATE {tracker.table_name} SET quantity=?, side=?, entry_price=? WHERE id=?",
                                (alp_qty, alp_side, alp_entry_price, sql_pos['id'])
                            )
                    else:
                        logger.info("Position %s exists on Alpaca but not in SQLite. Opening in SQLite: qty=%s, side=%s, entry=%s",
                                    sym, alp_qty, alp_side, alp_entry_price)
                        # Compute target price
                        tp_pct = float(tracker.config.get('trading', {}).get('take_profit_pct', 0.03))
                        if alp_side == 'SHORT':
                            target_price = alp_entry_price * (1.0 - tp_pct)
                        else:
                            target_price = alp_entry_price * (1.0 + tp_pct)

                        cursor.execute(
                            f"""
                            INSERT INTO {tracker.table_name} (
                                symbol, side, entry_date, entry_price, quantity, target_price, status, exit_reason
                            )
                            VALUES (?, ?, ?, ?, ?, ?, 'OPEN', 'alpaca_sync_open')
                            """,
                            (sym, alp_side, trade_date_str, alp_entry_price, alp_qty, target_price)
                        )

            logger.info("Synchronization of SQLite table %s complete.", tracker.table_name)
        except Exception as sync_exc:
            logger.error("Error synchronizing SQLite positions with Alpaca: %s", sync_exc, exc_info=True)

    def get_fallback_rankings_for_date_bulk(self, symbols, signal_date, conn, lookback_days=120):
        """
        Fallback ranking when no ML models are available.
        Uses simple 10-day momentum computed from price history.
        """
        sig = pd.to_datetime(signal_date)
        start = (sig - timedelta(days=int(lookback_days))).strftime("%Y-%m-%d")
        sig_str = sig.strftime("%Y-%m-%d")

        symbols = [str(s or "").strip().upper() for s in symbols if str(s or "").strip()]
        if not symbols:
            return []

        placeholders = ",".join(["?"] * len(symbols))
        q = (
            "SELECT symbol, date, close "
            "FROM prices "
            f"WHERE symbol IN ({placeholders}) AND date >= ? "
            "ORDER BY symbol, date"
        )
        df = pd.read_sql(q, conn, params=list(symbols) + [start])
        if df.empty:
            return []

        df["date"] = pd.to_datetime(df["date"])

        def _add_momentum(g):
            g = g.sort_values("date").reset_index(drop=True)
            g["return_10d"] = g["close"].pct_change(10)
            return g

        feat = df.groupby("symbol", group_keys=False).apply(_add_momentum)
        feat_on_date = feat[feat["date"] == pd.to_datetime(sig_str)]
        if feat_on_date.empty:
            return []

        rankings = []
        for _, row in feat_on_date.iterrows():
            sym = str(row.get("symbol", "")).strip().upper()
            if not sym:
                continue
            try:
                score = float(row.get("return_10d", 0.0))
            except Exception:
                continue
            rankings.append({"symbol": sym, "predicted_return": score})
        return rankings

    def compute_smc_signals_bulk(self, symbols, signal_date, conn, pivots_length=5):
        """
        Compute LuxAlgo SMC signals in bulk for a list of symbols on a given signal_date.
        """
        sig_date_str = pd.to_datetime(signal_date).strftime("%Y-%m-%d")
        start_date = (pd.to_datetime(signal_date) - timedelta(days=120)).strftime("%Y-%m-%d")
        
        symbols = [str(s).strip().upper() for s in symbols if str(s).strip()]
        if not symbols:
            return pd.DataFrame()
            
        placeholders = ",".join(["?"] * len(symbols))
        q = f"""
            SELECT symbol, date, open, high, low, close
            FROM prices
            WHERE symbol IN ({placeholders}) AND date <= ? AND date >= ?
            ORDER BY symbol, date ASC
        """
        df = pd.read_sql(q, conn, params=list(symbols) + [sig_date_str, start_date])
        if df.empty:
            return pd.DataFrame()
            
        df["date"] = pd.to_datetime(df["date"])
        L = pivots_length
        
        sig_rows = []
        
        for symbol, group in df.groupby("symbol"):
            group = group.sort_values("date").reset_index(drop=True)
            if len(group) < 2 * L + 1:
                continue
                
            # Vectorized rolling pivots
            group['is_pivot_high'] = (
                (group['high'] >= group['high'].shift(1).rolling(L).max()) & 
                (group['high'] > group['high'].shift(-L).rolling(L).max())
            )
            group['is_pivot_low'] = (
                (group['low'] <= group['low'].shift(1).rolling(L).min()) & 
                (group['low'] < group['low'].shift(-L).rolling(L).min())
            )
            
            # Chronological state loop
            trend_bias = "NEUTRAL"
            active_high_pivot = None
            active_low_pivot = None
            
            biases = []
            signals = []
            
            for i in range(len(group)):
                pivot_check_idx = i - L
                if pivot_check_idx >= 0:
                    if group.loc[pivot_check_idx, 'is_pivot_high']:
                        active_high_pivot = group.loc[pivot_check_idx, 'high']
                    if group.loc[pivot_check_idx, 'is_pivot_low']:
                        active_low_pivot = group.loc[pivot_check_idx, 'low']
                        
                close_t = group.loc[i, 'close']
                signal = "NONE"
                
                if trend_bias == "BEARISH":
                    if active_high_pivot is not None and close_t > active_high_pivot:
                        trend_bias = "BULLISH"
                        signal = "BULLISH_CHOCH"
                elif trend_bias == "BULLISH":
                    if active_low_pivot is not None and close_t < active_low_pivot:
                        trend_bias = "BEARISH"
                        signal = "BEARISH_CHOCH"
                elif trend_bias == "NEUTRAL":
                    if active_high_pivot is not None and close_t > active_high_pivot:
                        trend_bias = "BULLISH"
                        signal = "BULLISH_CHOCH"
                    elif active_low_pivot is not None and close_t < active_low_pivot:
                        trend_bias = "BEARISH"
                        signal = "BEARISH_CHOCH"
                        
                biases.append(trend_bias)
                signals.append(signal)
                
            group['trend_bias'] = biases
            group['smc_signal'] = signals
            group['return_10d'] = group['close'].pct_change(10)
            
            if group.iloc[-1]['date'].strftime("%Y-%m-%d") == sig_date_str:
                sig_rows.append(group.iloc[-1])
                
        return pd.DataFrame(sig_rows) if sig_rows else pd.DataFrame()

    def run_daily_test(self, test_date=None, pipeline_stats=None, backtest_signals=None):
        """
        Position-Tracking Backtest with SMC Strategy:
        1. Sync SQLite positions with Alpaca.
        2. Open new positions based on SMC CHOCH close-above-level signals.
        3. Check existing positions for 3% TP hits.
        4. Report unrealized P&L.
        """
        conn = sqlite3.connect(self.db_path)
        rank_df = pd.DataFrame()
        core_rank_df = pd.DataFrame()

        if test_date is None:
            last_date = pd.read_sql("SELECT MAX(date) as max_date FROM prices", conn).iloc[0, 0]
            if not last_date:
                logger.warning("No price data available for backtest.")
                conn.close()
                return None
            test_date = pd.to_datetime(last_date)
        else:
            test_date = pd.to_datetime(test_date)

        # signal_date is the day BEFORE test_date
        signal_date = test_date - timedelta(days=1)

        # Handle weekends/holidays to find the last available date in prices
        attempts = 0
        while attempts < 5:
            check = pd.read_sql(
                f"SELECT COUNT(*) as cnt FROM prices WHERE date='{signal_date.strftime('%Y-%m-%d')}'",
                conn
            )
            if check.iloc[0]['cnt'] > 0:
                break
            signal_date = signal_date - timedelta(days=1)
            attempts += 1

        logger.info(f"Running position-tracking backtest for {test_date.date()}")
        logger.info(f"Using signals from {signal_date.date()} (T-1)")

        trade_date_str = test_date.strftime('%Y-%m-%d')
        # Synchronize SQLite positions with Alpaca before doing any capital/portfolio checks
        if self.core_broker and not (str(os.getenv("DISABLE_CORE_TRADING", "")).strip().lower() in {"1", "true", "yes", "on"}):
            self._sync_sqlite_positions_with_alpaca(self.core_tracker, self.core_broker, trade_date_str)

        # Load universe
        universe_df = pd.read_csv(self.universe_path)

        # Always load the fallback momentum rankings into rank_df so that it contains the full momentum rankings
        rankings = self.get_fallback_rankings_for_date_bulk(
            universe_df["ticker"].tolist(),
            signal_date,
            conn,
        )
        if rankings:
            rank_df = pd.DataFrame(rankings)
            rank_df["rank"] = rank_df["predicted_return"].rank(ascending=False, method="first").astype(int)
            rank_df = rank_df.sort_values("rank").reset_index(drop=True)
            rank_df["penalty"] = 1.0
            rank_df["adjusted_score"] = rank_df["predicted_return"]
        ai_cfg = self.config.get("ai_trading", {})
        trading_cfg = self.config.get("trading", {})
        ai_disabled_by_env = str(os.getenv("DISABLE_AI_TRADING", "")).strip().lower() in {
            "1",
            "true",
            "yes",
            "on",
        }
        ai_enabled = bool(ai_cfg.get("enabled", False)) and not ai_disabled_by_env
        core_disabled_by_env = str(os.getenv("DISABLE_CORE_TRADING", "")).strip().lower() in {
            "1",
            "true",
            "yes",
            "on",
        }


        # Determine Current Capital (Compounding)
        initial_capital = float(self.config.get('trading', {}).get('initial_capital', 100000))
        leverage = float(self.config.get('trading', {}).get('leverage', 2))
        summary = self.core_tracker.get_portfolio_summary()
        total_realized_dollars = summary.get('total_realized_pnl_dollars', 0.0)
        total_unrealized_dollars = summary.get('total_unrealized_pnl_dollars', 0.0)
        current_capital = initial_capital + total_realized_dollars + total_unrealized_dollars
        capital_start = current_capital

        # Override current_capital with Alpaca equity when broker is connected
        if self.core_broker:
            try:
                alpaca_acct = self.core_broker.get_account()
                alpaca_equity = float(alpaca_acct.get('equity', current_capital))
                current_capital = alpaca_equity
                capital_start = current_capital
                logger.info("Position sizing: Alpaca equity=$%.2f", alpaca_equity)
            except Exception as e:
                logger.warning("Could not pull Alpaca equity for sizing, using SQLite: %s", e)

        # Get updated open positions and slots
        open_positions_df = self.core_tracker.get_open_positions()
        open_symbols = set(open_positions_df['symbol']) if not open_positions_df.empty else set()
        open_shorts_count = len(open_positions_df[open_positions_df['side'] == 'SHORT']) if not open_positions_df.empty else 0
        max_positions = self.config.get('trading', {}).get('max_positions', 10)
        max_shorts = self.config.get('trading', {}).get('max_shorts', 5)
        enable_shorts = bool(self.config.get("trading", {}).get("enable_shorts", False))
        
        available_slots = max(0, max_positions - len(open_symbols))
        invested_cost = 0.0
        if not open_positions_df.empty:
            invested_cost = (open_positions_df['entry_price'] * open_positions_df['quantity']).sum()

        if self.core_broker:
            try:
                alpaca_positions = self.core_broker.get_positions()
                invested_cost = sum(abs(float(p.get('market_value', 0))) for p in alpaca_positions)
                open_symbols = set(p.get('symbol', '') for p in alpaca_positions)
                available_slots = max(0, max_positions - len(open_symbols))
                open_shorts_count = sum(1 for p in alpaca_positions if str(p.get('side', '')).upper() == 'SHORT')
            except Exception:
                pass

        available_capital = max(0.0, current_capital - invested_cost)

        new_positions = []
        top_up_positions = []
        target_portfolio = pd.DataFrame()
        
        if not core_disabled_by_env and available_slots > 0 and available_capital > 0:
            # Generate SMC signals for the universe
            smc_all_df = self.compute_smc_signals_bulk(
                universe_df["ticker"].tolist(), signal_date, conn,
                pivots_length=self.config.get("smc", {}).get("pivots_length", 5)
            )
            
            candidates = []
            if not smc_all_df.empty:
                # Exclude currently open symbols
                non_held = smc_all_df[~smc_all_df['symbol'].isin(open_symbols)].copy()
                
                # Filter long and short candidates
                for _, row in non_held.iterrows():
                    sig = row['smc_signal']
                    sym = row['symbol']
                    ret10 = row['return_10d']
                    
                    if sig == "BULLISH_CHOCH":
                        candidates.append({
                            'symbol': sym,
                            'side': 'LONG',
                            'return_10d': ret10,
                            'momentum_score': ret10 if not pd.isna(ret10) else -999.0
                        })
                    elif enable_shorts and sig == "BEARISH_CHOCH":
                        candidates.append({
                            'symbol': sym,
                            'side': 'SHORT',
                            'return_10d': ret10,
                            'momentum_score': -ret10 if not pd.isna(ret10) else -999.0
                        })
            
            # Sort candidates by momentum_score descending
            candidates_df = pd.DataFrame(candidates)
            selected_entries = []
            
            if not candidates_df.empty:
                candidates_df = candidates_df.sort_values('momentum_score', ascending=False)
                
                # Select up to available slots, respecting max_shorts
                curr_shorts = open_shorts_count
                for _, cand in candidates_df.iterrows():
                    if len(selected_entries) >= available_slots:
                        break
                    if cand['side'] == 'SHORT':
                        if curr_shorts >= max_shorts:
                            continue
                        curr_shorts += 1
                    
                    selected_entries.append(cand)
            
            if selected_entries:
                target_portfolio = pd.DataFrame(selected_entries)
                target_portfolio['weight'] = 1.0 / len(target_portfolio)
                
                # Populate rank_lookup for logging/reasoning
                core_rank_df = target_portfolio.copy()
                core_rank_df['predicted_return'] = core_rank_df['return_10d']
                core_rank_df['adjusted_score'] = core_rank_df['momentum_score']
                core_rank_df['penalty'] = 1.0
                core_rank_df['rank'] = core_rank_df.index + 1
                rank_lookup = core_rank_df.set_index('symbol')[['predicted_return', 'adjusted_score', 'penalty', 'rank']]
            else:
                rank_lookup = pd.DataFrame(columns=['predicted_return', 'adjusted_score', 'penalty', 'rank'])
        else:
            rank_lookup = pd.DataFrame(columns=['predicted_return', 'adjusted_score', 'penalty', 'rank'])

        # Now execute entry orders for target_portfolio
        remaining_capital = float(available_capital)
        try:
            min_order_dollars_cfg = float(self.config.get("trading", {}).get("min_order_dollars", 500.0) or 500.0)
        except (TypeError, ValueError):
            min_order_dollars_cfg = 500.0
        try:
            min_order_equity_pct_cfg = float(self.config.get("trading", {}).get("min_order_equity_pct", 0.005) or 0.005)
        except (TypeError, ValueError):
            min_order_equity_pct_cfg = 0.005
        min_order_dollars = max(
            0.0,
            float(min_order_dollars_cfg),
            float(current_capital) * max(0.0, float(min_order_equity_pct_cfg)),
        )

        if not core_disabled_by_env and not target_portfolio.empty:
            for _, row in target_portfolio.iterrows():
                symbol = row['symbol']
                side = row['side']
                
                price_data = pd.read_sql(
                    "SELECT open FROM prices WHERE symbol=? AND date=?",
                    conn,
                    params=(symbol, trade_date_str)
                )
                
                entry_price = None
                if not price_data.empty:
                    val = price_data.iloc[0]['open']
                    if val is not None and not pd.isna(val):
                        entry_price = float(val)
                if (entry_price is None or pd.isna(entry_price)) and self.core_broker:
                    # Fall back to broker's real-time price
                    entry_price = self.core_broker.get_latest_price(symbol)
                
                if entry_price is not None and entry_price > 0.0:
                    requested_allocation = max(
                        0.0, float(row.get('weight', 0.0) or 0.0) * float(available_capital)
                    )
                    allocation_dollars = min(requested_allocation, max(0.0, remaining_capital))
                    if allocation_dollars <= 0.0:
                        continue
                    if allocation_dollars < min_order_dollars:
                        logger.info(
                            "Skipping %s %s entry because allocation %.2f is below minimum order floor %.2f.",
                            symbol,
                            side,
                            allocation_dollars,
                            min_order_dollars,
                        )
                    raw_qty = allocation_dollars / entry_price
                    qty = max(1, int(raw_qty))
                    quantity = float(qty)
                    allocation_pct = (allocation_dollars / current_capital * 100) if current_capital else 0.0
                    target_price = entry_price * 1.03 if side == "LONG" else entry_price * 0.97
                    
                    broker_order = None
                    if self.core_broker:
                        try:
                            alpaca_side = "buy" if side == "LONG" else "sell"
                            broker_order = self.core_broker.submit_order(
                                symbol=symbol,
                                qty=qty,
                                side=alpaca_side,
                                order_type="market",
                                time_in_force="day",
                                take_profit_price=target_price,
                            )
                            logger.info(
                                "Alpaca order submitted: %s %s %d shares @ market (order_id=%s, TP=%s)",
                                alpaca_side.upper(),
                                symbol,
                                qty,
                                broker_order.get("id", "unknown"),
                                f"${target_price:.2f}",
                            )
                        except AlpacaError as e:
                            logger.warning(
                                "Alpaca order rejected for %s: %s (code=%s). Skipping.",
                                symbol,
                                e.message,
                                e.code,
                            )
                            broker_order = None
                        except Exception as e:
                            logger.warning(
                                "Alpaca order failed for %s: %s. Skipping.",
                                symbol,
                                e,
                            )
                            broker_order = None
                            
                    # Record position in SQLite (only if simulation or broker order succeeded)
                    if (not self.core_broker) or (broker_order is not None):
                        pos_id = self.core_tracker.open_position(
                            symbol=symbol,
                            entry_date=trade_date_str,
                            entry_price=entry_price,
                            quantity=quantity,
                            side=side,
                            target_price=target_price,
                        )
                    else:
                        pos_id = None
                        
                    if pos_id:
                        reason = "SMC CHOCH Breakout"
                        if symbol in rank_lookup.index:
                            info = rank_lookup.loc[symbol]
                            pred = float(info['predicted_return'])
                            adj = float(info['adjusted_score'])
                            penalty = float(info['penalty'])
                            rank = int(info['rank'])
                            reason = (
                                f"Rank {rank} signal (pred {pred:.2%}, adj {adj:.2%}, penalty {penalty:.2f})"
                            )
                            
                        new_positions.append({
                            'symbol': symbol,
                            'side': side,
                            'entry_price': entry_price,
                            'target_price': target_price,
                            'quantity': quantity,
                            'allocation_pct': allocation_pct,
                            'allocation_dollars': allocation_dollars,
                            'reason': reason
                        })
                        remaining_capital = max(0.0, remaining_capital - allocation_dollars)

        entries_today = list(new_positions)

        # Keep Core + AI strategies distinct by optionally preventing overlap.
        core_reserved_symbols = set(open_symbols)
        if target_portfolio is not None and hasattr(target_portfolio, "empty") and not target_portfolio.empty:
            try:
                core_reserved_symbols |= set([str(s).strip().upper() for s in target_portfolio["symbol"].tolist() if str(s).strip()])
            except Exception:
                pass

        if rank_df.empty:
            rankings = self.get_fallback_rankings_for_date_bulk(
                universe_df["ticker"].tolist(),
                signal_date,
                conn,
            )
            if rankings:
                rank_df = pd.DataFrame(rankings)
                rank_df["rank"] = rank_df["predicted_return"].rank(ascending=False, method="first").astype(int)
                rank_df = rank_df.sort_values("rank").reset_index(drop=True)
                rank_df["penalty"] = 1.0
                rank_df["adjusted_score"] = rank_df["predicted_return"]

        conn.close()

        os.makedirs(self.results_dir, exist_ok=True)
        date_str = test_date.strftime('%Y%m%d')
        report = None
        unrealized = pd.DataFrame()
        closed_positions = []
        if core_disabled_by_env:
            logger.info("Core trading disabled by env; skipping core strategy path.")
        else:
            # Check ALL remaining open positions for TP hits on this day
            tp_closed_positions = self.core_tracker.check_and_close_positions(
                check_date=test_date.strftime('%Y-%m-%d')
            )

            # Close positions on Alpaca if broker is enabled
            if self.core_broker and tp_closed_positions:
                reverted_symbols = set()
                for pos in tp_closed_positions:
                    symbol = pos.get('symbol')
                    if symbol:
                        try:
                            self.core_broker.close_position(symbol)
                            logger.info("Alpaca position closed on TP: %s", symbol)
                        except AlpacaError as e:
                            logger.warning(
                                "Alpaca close failed for %s: %s (code=%s). Reverting database close.",
                                symbol,
                                e.message,
                                e.code,
                            )
                            self.core_tracker.reopen_position(symbol)
                            reverted_symbols.add(symbol)
                        except Exception as e:
                            logger.warning("Alpaca close failed for %s: %s. Reverting database close.", symbol, e)
                            self.core_tracker.reopen_position(symbol)
                            reverted_symbols.add(symbol)
                
                # Filter out the reverted positions
                if reverted_symbols:
                    tp_closed_positions = [p for p in tp_closed_positions if p.get('symbol') not in reverted_symbols]
            
            closed_positions = tp_closed_positions

            # Save closed trades for meta-learning (per-day)
            if closed_positions:
                trades_df = pd.DataFrame(closed_positions)
                trades_df = trades_df.rename(columns={
                    "realized_pnl": "strat_return",
                    "realized_pnl_dollars": "pnl_dollars",
                })
                trades_df.to_csv(
                    os.path.join(self.results_dir, f"trades_{date_str}.csv"),
                    index=False
                )

            # Get updated portfolio state
            summary = self.core_tracker.get_portfolio_summary()
            unrealized = self.core_tracker.get_unrealized_pnl()
            total_realized_dollars = summary.get('total_realized_pnl_dollars', 0.0)
            total_unrealized_dollars = summary.get('total_unrealized_pnl_dollars', 0.0)

            # Calculate Realized P&L for TODAY
            realized_today_dollars = sum([p.get('realized_pnl_dollars', 0.0) for p in closed_positions])
            realized_today = realized_today_dollars / 100000.0

            total_realized_pct = total_realized_dollars / 200000.0
            total_unrealized_pct = total_unrealized_dollars / 200000.0
            current_capital = initial_capital + total_realized_dollars + total_unrealized_dollars
            new_account_return = (current_capital - initial_capital) / 200000.0
            # Add previous account return
            previous_return_pct = float(self.config.get('trading', {}).get('previous_return_pct') or 0.0)
            total_account_return = (current_capital / 100000.0) / 2.0 - 0.5 + float(previous_return_pct or 0.0)

            # Report gross exposure while keeping cash side-aware. Shorts add exposure
            # but do not spend cash like long purchases, so they must not drive cash negative.
            open_positions_now = self.core_tracker.get_open_positions()
            invested_notional, available_cash = _position_cash_metrics(open_positions_now, current_capital)

            period_start_core = self.core_tracker.get_performance_period_start()
            report = {
                'date': test_date.date(),
                'performance_period_start': period_start_core,
                'new_positions_opened': len(new_positions),
                'positions_topped_up': len(top_up_positions),
                'positions_closed_at_tp': len(closed_positions),
                'open_positions': summary['open_positions'],
                'realized_pnl_today': realized_today,
                'realized_pnl_today_dollars': realized_today_dollars,
                'total_realized_pnl': total_realized_pct,
                'total_realized_pnl_dollars': total_realized_dollars,
                'total_unrealized_pnl': total_unrealized_pct,
                'total_unrealized_pnl_dollars': total_unrealized_dollars,
                'total_account_return': total_account_return,
                'current_capital_estimate': current_capital,
                'invested_notional': invested_notional,
                'available_cash': available_cash,
                'initial_capital': initial_capital,
                'previous_return_pct': float(previous_return_pct or 0.0)
            }

            # --- OVERRIDE ALL DATA with Alpaca when broker is connected ---
            if self.core_broker:
                try:
                    # Keep a copy of SQLite-based unrealized DataFrame before overriding
                    sqlite_unrealized = unrealized.copy() if not unrealized.empty else pd.DataFrame()

                    alpaca_positions = self.core_broker.get_positions()
                    alpaca_account = self.core_broker.get_account()

                    alpaca_equity = float(alpaca_account.get('equity', 0))
                    alpaca_cash = float(alpaca_account.get('cash', 0))

                    # --- Build unrealized DataFrame from Alpaca ---
                    alpaca_unrealized_rows = []
                    for p in alpaca_positions:
                        symbol = p.get('symbol', '')
                        side = str(p.get('side', 'long')).upper()
                        qty = float(p.get('qty', 0))
                        entry_price = float(p.get('avg_entry_price', 0))
                        current_price = float(p.get('current_price', 0))
                        unrealized_pl = float(p.get('unrealized_pl', 0))
                        unrealized_plpc = float(p.get('unrealized_plpc', 0))

                        entry_date = None
                        target_price = None
                        if not sqlite_unrealized.empty:
                            match = sqlite_unrealized[sqlite_unrealized['symbol'] == symbol]
                            if not match.empty:
                                entry_date = match.iloc[0].get('entry_date')
                                target_price = match.iloc[0].get('target_price')

                        alpaca_unrealized_rows.append({
                            'symbol': symbol,
                            'side': side,
                            'entry_date': entry_date,
                            'entry_price': entry_price,
                            'quantity': qty,
                            'current_price': current_price,
                            'target_price': target_price,
                            'unrealized_pnl': unrealized_plpc,
                            'unrealized_pnl_dollars': unrealized_pl,
                            'distance_to_tp': abs(target_price - current_price) / current_price if target_price and current_price else None,
                        })

                    # Add any positions from SQLite that are not in Alpaca positions (e.g. newly entered today or historical)
                    alpaca_symbols = {p.get('symbol') for p in alpaca_positions}
                    if not sqlite_unrealized.empty:
                        for _, row in sqlite_unrealized.iterrows():
                            sym = row.get('symbol')
                            if sym and sym not in alpaca_symbols:
                                alpaca_unrealized_rows.append({
                                    'symbol': sym,
                                    'side': row.get('side'),
                                    'entry_date': row.get('entry_date'),
                                    'entry_price': row.get('entry_price'),
                                    'quantity': row.get('quantity'),
                                    'current_price': row.get('current_price'),
                                    'target_price': row.get('target_price'),
                                    'unrealized_pnl': row.get('unrealized_pnl'),
                                    'unrealized_pnl_dollars': row.get('unrealized_pnl_dollars'),
                                    'distance_to_tp': row.get('distance_to_tp'),
                                })

                    unrealized = pd.DataFrame(alpaca_unrealized_rows)

                    # --- Recompute P&L and metrics from the final combined list of open positions ---
                    alpaca_invested = sum(abs(float(row['quantity'] or 0) * float(row['current_price'] or 0)) for row in alpaca_unrealized_rows)
                    alpaca_unrealized_dollars = sum(float(row['unrealized_pnl_dollars'] or 0) for row in alpaca_unrealized_rows)

                    # --- Realized P&L: equity = initial + realized + unrealized ---
                    alpaca_realized_dollars = alpaca_equity - initial_capital - alpaca_unrealized_dollars

                    # --- Return calculation ---
                    equity_at_reset = self.config.get('trading', {}).get('equity_at_reset')
                    if equity_at_reset is None:
                        equity_at_reset = initial_capital + float(previous_return_pct or 0.0) * 200000.0
                    else:
                        equity_at_reset = float(equity_at_reset)
                    new_account_return = (alpaca_equity - equity_at_reset) / 200000.0
                    total_account_return = (alpaca_equity / 100000.0) / 2.0 - 0.5 + float(previous_return_pct or 0.0)

                    # --- Override report dict ---
                    report['open_positions'] = len(unrealized)
                    report['current_capital_estimate'] = alpaca_equity
                    report['invested_notional'] = alpaca_invested
                    # Calculate available cash as equity minus long positions value to satisfy Accounting Identity
                    alpaca_long_notional = sum(
                        float(row['quantity'] or 0) * float(row['current_price'] or 0)
                        for row in alpaca_unrealized_rows
                        if str(row.get('side', 'LONG')).upper() != 'SHORT'
                    )
                    report['available_cash'] = alpaca_equity - alpaca_long_notional
                    report['total_realized_pnl'] = alpaca_realized_dollars / 200000.0
                    report['total_realized_pnl_dollars'] = alpaca_realized_dollars
                    report['total_unrealized_pnl'] = alpaca_unrealized_dollars / 200000.0
                    report['total_unrealized_pnl_dollars'] = alpaca_unrealized_dollars
                    report['total_account_return'] = total_account_return
                    report['previous_return_pct'] = float(previous_return_pct or 0.0)

                    # --- Pull today's orders from Alpaca for entered/closed tables ---
                    from datetime import date as _date
                    today_str = str(test_date.date()) if hasattr(test_date, 'date') else str(test_date)
                    all_orders = self.core_broker.get_orders(status='all', limit=200)

                    # Deduplicate by order id
                    seen_ids = set()
                    unique_orders = []
                    for o in all_orders:
                        oid = o.get('id')
                        if oid and oid not in seen_ids:
                            seen_ids.add(oid)
                            unique_orders.append(o)

                    alpaca_new_positions = []
                    alpaca_closed_positions = []
                    alpaca_realized_today_dollars = 0.0

                    for o in unique_orders:
                        created_at = o.get('created_at', '')
                        filled_at = o.get('filled_at', '')
                        is_today = (created_at and str(created_at)[:10] == today_str) or (filled_at and str(filled_at)[:10] == today_str)
                        if not is_today:
                            continue

                        symbol = o.get('symbol', '')
                        side = o.get('side', '')
                        status = o.get('status', '').lower()
                        if status in ('canceled', 'rejected', 'expired'):
                            continue

                        is_filled = (status == 'filled')
                        if not is_filled:
                            continue
                        qty = float(o.get('qty', 0) or 0)
                        qty_abs = abs(qty)
                        filled_avg_price = float(o.get('filled_avg_price', 0) or 0)

                        # Find matching Alpaca position
                        alpaca_pos = None
                        for p in alpaca_positions:
                            if p.get('symbol') == symbol:
                                alpaca_pos = p
                                break

                        # Correct opening/closing determination
                        is_opening = False
                        is_closing = False
                        if side == 'buy':
                            if alpaca_pos and str(alpaca_pos.get('side', '')).lower() == 'short':
                                is_closing = True
                            else:
                                is_opening = True
                        elif side == 'sell':
                            if alpaca_pos and str(alpaca_pos.get('side', '')).lower() == 'long':
                                is_closing = True
                            else:
                                is_opening = True

                        local_match = None
                        for p in entries_today:
                            if p.get('symbol') == symbol:
                                local_match = p
                                break

                        if is_opening:
                            target_price = None
                            if local_match:
                                target_price = local_match.get('target_price')
                                notional = local_match.get('allocation_dollars', 0.0)
                                alloc_pct = local_match.get('allocation_pct', 0.0)
                                entry_p = filled_avg_price if is_filled else local_match.get('entry_price')
                                qty_abs = local_match.get('quantity', qty_abs)
                                reason = local_match.get('reason')
                            else:
                                if not unrealized.empty:
                                    m = unrealized[unrealized['symbol'] == symbol]
                                    if not m.empty:
                                        target_price = m.iloc[0].get('target_price')
                                entry_p = filled_avg_price if is_filled else 0.0
                                notional = entry_p * qty_abs
                                alloc_pct = (notional / alpaca_equity * 100) if alpaca_equity else 0.0
                                reason = "Core target portfolio entry"

                            side_label = 'LONG' if side == 'buy' else 'SHORT'
                            if not is_filled:
                                side_label += ' (Pending)'

                            alpaca_new_positions.append({
                                'symbol': symbol,
                                'side': side_label,
                                'entry_price': entry_p if is_filled else "Pending",
                                'target_price': target_price,
                                'allocation_pct': alloc_pct,
                                'allocation_dollars': notional,
                                'quantity': qty_abs,
                                'reason': reason,
                            })
                        elif is_closing:
                            realized_pct = 0.0
                            realized_dollars = 0.0
                            entry_p = 0.0
                            exit_p = filled_avg_price if is_filled else 0.0

                            local_close_match = None
                            for p in closed_positions:
                                if p.get('symbol') == symbol:
                                    local_close_match = p
                                    break

                            if alpaca_pos:
                                entry_p = float(alpaca_pos.get('avg_entry_price', 0))
                                ps = str(alpaca_pos.get('side', '')).lower()
                                if is_filled:
                                    if ps == 'long':
                                        realized_dollars = (exit_p - entry_p) * qty_abs
                                    else:
                                        realized_dollars = (entry_p - exit_p) * qty_abs
                                    realized_pct = realized_dollars / (entry_p * qty_abs) if entry_p and qty_abs else 0.0
                            elif local_close_match:
                                entry_p = local_close_match.get('entry_price', 0.0)
                                if is_filled:
                                    realized_dollars = local_close_match.get('realized_pnl_dollars', 0.0)
                                    realized_pct = local_close_match.get('realized_pnl', 0.0)

                            alpaca_realized_today_dollars += realized_dollars
                            side_label = 'LONG' if side == 'sell' else 'SHORT'
                            if not is_filled:
                                side_label += ' (Pending Exit)'

                            alpaca_closed_positions.append({
                                'symbol': symbol,
                                'side': side_label,
                                'entry_price': entry_p,
                                'exit_price': exit_p if is_filled else "Pending",
                                'realized_pnl': realized_pct if is_filled else 0.0,
                                'realized_pnl_dollars': realized_dollars if is_filled else 0.0,
                                'reason': local_close_match.get('reason') if local_close_match else "Core rotation exit",
                            })

                    # Override entries_today and closed_positions
                    entries_today = alpaca_new_positions
                    closed_positions = alpaca_closed_positions
                    report['positions_closed_at_tp'] = len(alpaca_closed_positions)
                    report['new_positions_opened'] = len(alpaca_new_positions)
                    report['realized_pnl_today_dollars'] = alpaca_realized_today_dollars
                    report['realized_pnl_today'] = alpaca_realized_today_dollars / 200000.0

                    logger.info(
                        "Alpaca FULL override: equity=$%.2f, realized=$%.2f, unrealized=$%.2f, return=%.2f%%, entered=%d, closed=%d",
                        alpaca_equity, alpaca_realized_dollars, alpaca_unrealized_dollars,
                        total_account_return * 100, len(alpaca_new_positions), len(alpaca_closed_positions),
                    )
                except Exception as e:
                    logger.warning("Failed to pull Alpaca data for report, using SQLite: %s", e)
                    import traceback
                    logger.warning(traceback.format_exc())

            pd.DataFrame([report]).to_csv(
                os.path.join(self.results_dir, f"daily_report_{date_str}.csv"),
                index=False
            )

            if not unrealized.empty:
                unrealized.to_csv(
                    os.path.join(self.results_dir, f"unrealized_{date_str}.csv"),
                    index=False
                )

            logger.info(f"Daily report saved to {self.results_dir}")

        # --- AI strategy (separate $100k account) ---
        from llm_trader import propose_trades_with_llm

        if ai_disabled_by_env:
            ai_enabled = False
        ai_report = None
        ai_unrealized = pd.DataFrame()
        ai_closed = []
        ai_new = []
        ai_llm_status = {"enabled": False, "ok": False, "error": "disabled"}
        if ai_disabled_by_env:
            ai_llm_status = {"enabled": False, "ok": True, "error": "disabled_by_env"}

        if ai_enabled:
            ai_initial_capital = float(ai_cfg.get("initial_capital", 100000))
            ai_max_positions = int(ai_cfg.get("max_positions", 10))
            ai_allow_shorts = bool(ai_cfg.get("allow_shorts", True))
            ai_max_shorts = int(ai_cfg.get("max_shorts", 5))
            ai_position_management_mode = str(
                ai_cfg.get("position_management_mode", "autonomous_rebalance") or "autonomous_rebalance"
            ).strip().lower()
            ai_min_trade_dollars = float(
                ai_cfg.get("min_trade_dollars", trading_cfg.get("min_order_dollars", 500)) or 0.0
            )
            # Slippage model: percentage of price added as realistic execution cost
            ai_slippage_pct = float(ai_cfg.get("slippage_pct", 0.0005) or 0.0005)
            ai_previous_return_pct = float(ai_cfg.get('previous_return_pct') or 0.0)
            ai_leverage = float(ai_cfg.get('leverage', 2))

            def _pyfloat(value):
                try:
                    if pd.isna(value):
                        return None
                    return float(value)
                except Exception:
                    return None

            def _recent_price_metrics(conn_, sym_, lookback=30):
                dfp = pd.read_sql(
                    "SELECT date, close, volume FROM prices WHERE symbol=? ORDER BY date DESC LIMIT ?",
                    conn_,
                    params=(sym_, int(lookback)),
                )
                if dfp.empty or len(dfp) < 2:
                    return None
                dfp = dfp.sort_values("date").reset_index(drop=True)
                closes = dfp["close"].astype(float).tolist()
                tail_n = min(10, len(closes))
                closes_tail = [float(x) for x in closes[-tail_n:]]
                v20 = float(dfp["volume"].astype(float).tail(20).mean()) if "volume" in dfp.columns else None
                v1 = float(dfp["volume"].astype(float).iloc[-1]) if "volume" in dfp.columns else None
                return {
                    "last_date": str(dfp["date"].iloc[-1]),
                    "last_close": float(closes_tail[-1]),
                    "closes_tail": closes_tail,
                    "volume_1d": v1,
                    "volume_20d_avg": v20,
                }

            def _latest_feature_snapshot(conn_, sym_, lookback=80, news_window_days=7):
                dfp = pd.read_sql(
                    "SELECT date, close, volume FROM prices WHERE symbol=? ORDER BY date DESC LIMIT ?",
                    conn_,
                    params=(sym_, int(lookback)),
                )
                if dfp.empty or len(dfp) < 20:
                    return {
                        "return_1d": None,
                        "return_5d": None,
                        "return_10d": None,
                        "volatility_20d": None,
                        "dist_ma_20": None,
                        "dist_ma_50": None,
                        "rsi_14": None,
                        "volume_ratio": None,
                        "news_count_7d": 0,
                        "news_sentiment_7d": 0.0,
                    }

                dfp = dfp.sort_values("date").reset_index(drop=True).copy()
                dfp["close"] = pd.to_numeric(dfp["close"], errors="coerce")
                dfp["volume"] = pd.to_numeric(dfp["volume"], errors="coerce")
                dfp["return_1d"] = dfp["close"].pct_change(1)
                dfp["return_5d"] = dfp["close"].pct_change(5)
                dfp["return_10d"] = dfp["close"].pct_change(10)
                dfp["volatility_20d"] = dfp["return_1d"].rolling(20).std()
                dfp["ma_20"] = dfp["close"].rolling(20).mean()
                dfp["ma_50"] = dfp["close"].rolling(50).mean()
                dfp["dist_ma_20"] = (dfp["close"] - dfp["ma_20"]) / dfp["ma_20"]
                dfp["dist_ma_50"] = (dfp["close"] - dfp["ma_50"]) / dfp["ma_50"]

                delta = dfp["close"].diff()
                gain = delta.where(delta > 0, 0.0).rolling(window=14).mean()
                loss = (-delta.where(delta < 0, 0.0)).rolling(window=14).mean()
                rs = gain / loss.replace(0, pd.NA)
                dfp["rsi_14"] = 100 - (100 / (1 + rs))
                dfp["volume_ma_20"] = dfp["volume"].rolling(20).mean()
                dfp["volume_ratio"] = dfp["volume"] / dfp["volume_ma_20"]

                latest = dfp.iloc[-1]
                end_dt = pd.to_datetime(latest["date"])
                news_count_7d = 0
                news_sentiment_7d = 0.0
                try:
                    news_df = pd.read_sql(
                        "SELECT datetime, sentiment_score FROM news WHERE symbol=? ORDER BY datetime DESC LIMIT 250",
                        conn_,
                        params=(sym_,),
                    )
                    if not news_df.empty and "datetime" in news_df.columns:
                        news_df["datetime"] = pd.to_datetime(news_df["datetime"], errors="coerce")
                        news_df = news_df.dropna(subset=["datetime"])
                        if not news_df.empty:
                            start_dt = end_dt - pd.Timedelta(days=int(news_window_days or 7))
                            news_df = news_df[news_df["datetime"] >= start_dt]
                            news_count_7d = int(len(news_df))
                            if "sentiment_score" in news_df.columns:
                                sentiment = pd.to_numeric(news_df["sentiment_score"], errors="coerce").dropna()
                                if not sentiment.empty:
                                    news_sentiment_7d = float(sentiment.mean())
                except Exception:
                    news_count_7d = 0
                    news_sentiment_7d = 0.0

                return {
                    "return_1d": _pyfloat(latest.get("return_1d")),
                    "return_5d": _pyfloat(latest.get("return_5d")),
                    "return_10d": _pyfloat(latest.get("return_10d")),
                    "volatility_20d": _pyfloat(latest.get("volatility_20d")),
                    "dist_ma_20": _pyfloat(latest.get("dist_ma_20")),
                    "dist_ma_50": _pyfloat(latest.get("dist_ma_50")),
                    "rsi_14": _pyfloat(latest.get("rsi_14")),
                    "volume_ratio": _pyfloat(latest.get("volume_ratio")),
                    "news_count_7d": int(news_count_7d),
                    "news_sentiment_7d": float(news_sentiment_7d),
                }

            def _symbol_set(df_):
                if df_ is None or not hasattr(df_, "empty") or df_.empty or "symbol" not in df_.columns:
                    return set()
                return {
                    str(symbol).strip().upper()
                    for symbol in df_["symbol"].tolist()
                    if str(symbol).strip()
                }

            def _price_rows_for_date(conn_, symbols_, trade_date_str_):
                normalized_symbols = [str(sym).strip().upper() for sym in list(symbols_ or []) if str(sym).strip()]
                if not normalized_symbols:
                    return {}
                placeholders = ",".join(["?"] * len(normalized_symbols))
                df_prices = pd.read_sql(
                    f"SELECT symbol, date, open, close FROM prices WHERE date=? AND symbol IN ({placeholders})",
                    conn_,
                    params=[trade_date_str_] + normalized_symbols,
                )
                rows = {}
                if df_prices.empty:
                    return rows
                for _, row in df_prices.iterrows():
                    symbol = str(row.get("symbol") or "").strip().upper()
                    if not symbol:
                        continue
                    rows[symbol] = {
                        "date": str(row.get("date") or trade_date_str_),
                        "open": _pyfloat(row.get("open")),
                        "close": _pyfloat(row.get("close")),
                    }
                return rows

            def _market_exposure_from_open_positions(df_, price_rows_):
                if df_ is None or not hasattr(df_, "empty") or df_.empty:
                    return 0.0
                exposure = 0.0
                for _, pos in df_.iterrows():
                    symbol = str(pos.get("symbol") or "").strip().upper()
                    if not symbol:
                        continue
                    price_row = price_rows_.get(symbol) or {}
                    market_price = _pyfloat(price_row.get("open"))
                    if market_price is None:
                        market_price = _pyfloat(price_row.get("close"))
                    if market_price is None:
                        market_price = _pyfloat(pos.get("entry_price"))
                    quantity = _pyfloat(pos.get("quantity")) or 0.0
                    exposure += max(0.0, float(market_price) * float(quantity))
                return float(exposure)

            def _ranked_ai_candidate_symbols(
                symbols_,
                *,
                rank_df_,
                prompt_limit_,
                allow_shorts_,
                max_shorts_,
                seed_,
            ):
                """
                Keep the expensive trained-model CPU pass bounded without making
                candidates random. Price ingestion still scans the full universe;
                this only orders the subset that gets sent to the trained model.
                """
                normalized = []
                seen = set()
                for sym in list(symbols_ or []):
                    symbol = str(sym).strip().upper()
                    if symbol and symbol not in seen:
                        seen.add(symbol)
                        normalized.append(symbol)

                if not normalized:
                    return []

                prompt_slots = max(1, int(prompt_limit_ or 1))
                selected = []
                selected_seen = set()

                def add_symbol(value):
                    symbol = str(value).strip().upper()
                    if symbol and symbol in seen and symbol not in selected_seen:
                        selected_seen.add(symbol)
                        selected.append(symbol)

                rank_for_ai = pd.DataFrame()
                if rank_df_ is not None and hasattr(rank_df_, "empty") and not rank_df_.empty:
                    rank_for_ai = rank_df_.copy()
                    if "symbol" in rank_for_ai.columns and "adjusted_score" in rank_for_ai.columns:
                        rank_for_ai["symbol"] = rank_for_ai["symbol"].astype(str).str.strip().str.upper()
                        rank_for_ai = rank_for_ai[rank_for_ai["symbol"].isin(seen)].copy()
                        rank_for_ai["adjusted_score"] = pd.to_numeric(
                            rank_for_ai["adjusted_score"], errors="coerce"
                        ).fillna(0.0)
                    else:
                        rank_for_ai = pd.DataFrame()

                if not rank_for_ai.empty:
                    short_slots = 0
                    if bool(allow_shorts_) and int(max_shorts_ or 0) > 0 and prompt_slots > 1:
                        short_slots = min(
                            max(1, prompt_slots // 3, int(max_shorts_ or 0) * 2),
                            max(1, prompt_slots // 2),
                        )
                    long_slots = max(0, prompt_slots - short_slots)

                    for symbol in rank_for_ai.sort_values("adjusted_score", ascending=False)["symbol"].head(long_slots):
                        add_symbol(symbol)

                    if short_slots > 0:
                        for symbol in rank_for_ai.sort_values("adjusted_score", ascending=True)["symbol"].head(short_slots):
                            add_symbol(symbol)

                    rank_for_ai["_abs_adjusted_score"] = rank_for_ai["adjusted_score"].abs()
                    for symbol in rank_for_ai.sort_values("_abs_adjusted_score", ascending=False)["symbol"]:
                        if len(selected) >= prompt_slots:
                            break
                        add_symbol(symbol)

                # Deterministic random tail only fills missing slots after the
                # ranked high-conviction candidates are included.
                tail = [sym for sym in normalized if sym not in selected_seen]
                random.Random(seed_).shuffle(tail)
                return selected + tail

            trade_date_str = test_date.strftime('%Y-%m-%d')
            # Synchronize SQLite AI positions with Alpaca before doing any capital/portfolio checks
            if self.ai_broker:
                self._sync_sqlite_positions_with_alpaca(self.ai_tracker, self.ai_broker, trade_date_str)
            ai_summary_pre_actions = self.ai_tracker.get_portfolio_summary()
            ai_realized_pre_actions = float(ai_summary_pre_actions.get("total_realized_pnl_dollars", 0.0))
            ai_capital_pre_actions = ai_initial_capital + ai_realized_pre_actions
            ai_open_df_pre = self.ai_tracker.get_open_positions()
            ai_open_symbols_pre = _symbol_set(ai_open_df_pre)
            ai_topups = []

            universe_symbols = [str(t).strip().upper() for t in universe_df["ticker"].tolist() if str(t).strip()]
            priority_symbols = sorted(set(ai_open_symbols_pre))
            candidate_symbols = [s for s in universe_symbols if s not in ai_open_symbols_pre]

            disallow_overlap = bool(ai_cfg.get("disallow_core_overlap", True))
            blocked_by_core = 0
            if disallow_overlap and core_reserved_symbols:
                before_symbols = set(candidate_symbols)
                candidate_symbols = [s for s in candidate_symbols if s not in core_reserved_symbols]
                blocked_by_core = len(before_symbols) - len(candidate_symbols)

            conn_ai = sqlite3.connect(self.db_path)
            cand = []
            ai_trades = []
            price_rows = {}
            try:
                seed = f"{pd.to_datetime(signal_date).date().isoformat()}-ai"

                prompt_limit_cfg = int(ai_cfg.get("prompt_candidates_limit", 200) or 200)
                prompt_limit_env_raw = str(os.getenv("AI_PROMPT_CANDIDATES_LIMIT") or "").strip()
                if prompt_limit_env_raw:
                    try:
                        prompt_limit = int(float(prompt_limit_env_raw))
                    except Exception:
                        prompt_limit = prompt_limit_cfg
                else:
                    prompt_limit = prompt_limit_cfg
                prompt_limit = max(1, prompt_limit)
                candidate_symbols = _ranked_ai_candidate_symbols(
                    candidate_symbols,
                    rank_df_=rank_df,
                    prompt_limit_=prompt_limit,
                    allow_shorts_=ai_allow_shorts,
                    max_shorts_=ai_max_shorts,
                    seed_=seed,
                )
                effective_prompt_limit = max(prompt_limit, len(priority_symbols))
                price_lookback = int(ai_cfg.get("price_lookback_days", 30) or 30)
                feature_lookback = int(ai_cfg.get("feature_lookback_days", 80) or 80)
                news_window_days = int(ai_cfg.get("news_window_days", 7) or 7)

                open_lookup = {}
                if ai_open_df_pre is not None and hasattr(ai_open_df_pre, "empty") and not ai_open_df_pre.empty:
                    for _, pos in ai_open_df_pre.iterrows():
                        symbol = str(pos.get("symbol") or "").strip().upper()
                        if symbol and symbol not in open_lookup:
                            open_lookup[symbol] = pos

                for sym in list(priority_symbols) + list(candidate_symbols):
                    if len(cand) >= max(1, effective_prompt_limit):
                        break
                    snapshot = _recent_price_metrics(conn_ai, sym, lookback=price_lookback)
                    if not snapshot:
                        continue
                    snapshot.update(
                        _latest_feature_snapshot(
                            conn_ai,
                            sym,
                            lookback=feature_lookback,
                            news_window_days=news_window_days,
                        )
                    )
                    current_pos = open_lookup.get(sym)
                    if current_pos is not None:
                        snapshot["current_position_side"] = str(current_pos.get("side") or "LONG").upper()
                        snapshot["current_position_entry"] = _pyfloat(current_pos.get("entry_price"))
                        snapshot["current_position_qty"] = _pyfloat(current_pos.get("quantity"))
                    snapshot["symbol"] = sym
                    snapshot["as_of_date"] = str(pd.to_datetime(signal_date).date())
                    cand.append(snapshot)

                ai_trades, ai_llm_status = propose_trades_with_llm(
                    self.config,
                    cand,
                    max_positions=ai_max_positions,
                    allow_shorts=ai_allow_shorts,
                    max_shorts=ai_max_shorts,
                )
                target_symbols = {
                    str(trade.get("symbol") or "").strip().upper()
                    for trade in ai_trades
                    if str(trade.get("symbol") or "").strip()
                }
                execution_symbols = sorted(set(ai_open_symbols_pre) | target_symbols)
                price_rows = _price_rows_for_date(conn_ai, execution_symbols, trade_date_str)
            finally:
                conn_ai.close()

            if isinstance(ai_llm_status, dict):
                ai_llm_status["manager_mode"] = ai_position_management_mode
                ai_llm_status["disallow_core_overlap"] = disallow_overlap
                ai_llm_status["blocked_by_core"] = blocked_by_core
                ai_llm_status["candidates_built"] = len(cand)
                ai_llm_status["positions_evaluated"] = len(ai_open_symbols_pre)

            if not (isinstance(ai_llm_status, dict) and ai_llm_status.get("ok")):
                ai_trades = []
                if isinstance(ai_llm_status, dict):
                    ai_llm_status["entries_blocked_due_to_llm_error"] = True
            else:
                skipped_reason = str((ai_llm_status or {}).get("skipped_reason") or "").strip().lower()
                suppress_rotation_closes = skipped_reason in {"all_neutral", "no_action"} or bool(
                    (ai_llm_status or {}).get("entries_blocked_due_to_llm_error")
                )
                target_map = {
                    str(trade.get("symbol") or "").strip().upper(): trade
                    for trade in ai_trades
                    if str(trade.get("symbol") or "").strip()
                }
                current_positions = {}
                if ai_open_df_pre is not None and hasattr(ai_open_df_pre, "empty") and not ai_open_df_pre.empty:
                    for _, pos in ai_open_df_pre.iterrows():
                        symbol = str(pos.get("symbol") or "").strip().upper()
                        if symbol and symbol not in current_positions:
                            current_positions[symbol] = pos

                # Minimum holding period: skip closing positions opened today
                ai_min_holding_days = int(ai_cfg.get("min_holding_days", 1) or 1)
                today_str = str(test_date.strftime("%Y-%m-%d"))

                for symbol, pos in current_positions.items():
                    current_side = str(pos.get("side") or "LONG").upper()
                    target = target_map.get(symbol)
                    exec_row = price_rows.get(symbol) or {}
                    exec_price_raw = _pyfloat(exec_row.get("open"))
                    if exec_price_raw is None:
                        exec_price_raw = _pyfloat(exec_row.get("close"))
                    if exec_price_raw is None and self.ai_broker:
                        exec_price_raw = self.ai_broker.get_latest_price(symbol)
                    # Apply slippage for closing: selling longs gets worse price, buying shorts gets worse price
                    if exec_price_raw is not None and ai_slippage_pct > 0:
                        if current_side == "LONG":
                            exec_price = exec_price_raw * (1 - ai_slippage_pct)
                        else:
                            exec_price = exec_price_raw * (1 + ai_slippage_pct)
                    else:
                        exec_price = exec_price_raw
                    if target is None or str(target.get("side") or "LONG").upper() != current_side:
                        if suppress_rotation_closes and target is None:
                            self.ai_tracker.update_position_decision(
                                symbol,
                                trade_date_str,
                                decision_label="HOLD",
                                decision_confidence=None,
                                decision_reason=(
                                    "AI rotation close suppressed because manager produced no actionable targets "
                                    f"(skipped_reason={skipped_reason or 'none'})."
                                ),
                                target_price=_pyfloat(pos.get("entry_price")),
                            )
                            continue
                        if exec_price is None or exec_price <= 0.0:
                            _record_pipeline_issue(
                                pipeline_stats,
                                "WARNING",
                                f"AI Execution:{symbol}",
                                "Missing execution price for model-driven close; keeping position open.",
                            )
                            continue
                        # Minimum holding period: skip closing positions opened today
                        entry_date_str = str(pos.get("entry_date") or "")
                        if entry_date_str and ai_min_holding_days > 0:
                            try:
                                entry_dt = datetime.strptime(entry_date_str[:10], "%Y-%m-%d")
                                current_dt = datetime.strptime(today_str[:10], "%Y-%m-%d")
                                days_held = (current_dt - entry_dt).days
                                if days_held < ai_min_holding_days:
                                    self.ai_tracker.update_position_decision(
                                        symbol,
                                        trade_date_str,
                                        decision_label="HOLD",
                                        decision_confidence=None,
                                        decision_reason=(
                                            f"AI rotation close skipped: minimum holding period ({ai_min_holding_days}d) "
                                            f"not met ({days_held}d held)."
                                        ),
                                        target_price=_pyfloat(pos.get("entry_price")),
                                    )
                                    continue
                            except (TypeError, ValueError, AttributeError):
                                pass
                        reason = "AI rotation: symbol removed from target portfolio"
                        if target is not None:
                            reason = f"AI rotation: side changed {current_side}->{str(target.get('side') or 'LONG').upper()}"
                        closed = self.ai_tracker.close_position(
                            symbol=symbol,
                            exit_date=trade_date_str,
                            exit_price=exec_price,
                            reason=reason,
                        )

                        # Close position on Alpaca if broker is enabled
                        submit_success = True
                        if closed and self.ai_broker:
                            try:
                                self.ai_broker.close_position(symbol)
                                logger.info("Alpaca AI position closed: %s", symbol)
                            except AlpacaError as e:
                                logger.warning(
                                    "Alpaca AI close failed for %s: %s (code=%s). Reverting database close.",
                                    symbol,
                                    e.message,
                                    e.code,
                                )
                                self.ai_tracker.reopen_position(symbol)
                                submit_success = False
                            except Exception as e:
                                logger.warning("Alpaca AI close failed for %s: %s. Reverting database close.", symbol, e)
                                self.ai_tracker.reopen_position(symbol)
                                submit_success = False

                        if closed and submit_success:
                            ai_closed.append(closed)
                    else:
                        self.ai_tracker.update_position_decision(
                            symbol,
                            trade_date_str,
                            decision_label=target.get("label"),
                            decision_confidence=target.get("confidence"),
                            decision_reason=target.get("reason"),
                            target_price=_pyfloat(pos.get("entry_price")),
                        )

                ai_open_df_live = self.ai_tracker.get_open_positions()
                live_lookup = {}
                if ai_open_df_live is not None and hasattr(ai_open_df_live, "empty") and not ai_open_df_live.empty:
                    for _, pos in ai_open_df_live.iterrows():
                        symbol = str(pos.get("symbol") or "").strip().upper()
                        if symbol and symbol not in live_lookup:
                            live_lookup[symbol] = pos

                ai_summary_live = self.ai_tracker.get_portfolio_summary()
                ai_realized_live = float(ai_summary_live.get("total_realized_pnl_dollars", 0.0))
                ai_unrealized_live = float(ai_summary_live.get("total_unrealized_pnl_dollars", 0.0))
                ai_equity_live = ai_initial_capital + ai_realized_live + ai_unrealized_live

                # Override with Alpaca equity when broker is connected (same as Core)
                if self.ai_broker:
                    try:
                        ai_acct_live = self.ai_broker.get_account()
                        ai_alpaca_equity_live = float(ai_acct_live.get('equity', ai_equity_live))
                        ai_equity_live = ai_alpaca_equity_live
                        logger.info("AI position sizing: Alpaca equity=$%.2f, leverage=%.1fx", ai_alpaca_equity_live, ai_leverage)
                    except Exception as e:
                        logger.warning("Could not pull Alpaca AI equity for sizing, using SQLite: %s", e)

                ai_market_exposure = _market_exposure_from_open_positions(ai_open_df_live, price_rows)
                ai_available_capital = max(0.0, ai_equity_live - ai_market_exposure)

                for trade in sorted(ai_trades, key=lambda item: float(item.get("weight", 0.0) or 0.0), reverse=True):
                    symbol = str(trade.get("symbol") or "").strip().upper()
                    if not symbol:
                        continue
                    side = str(trade.get("side") or "LONG").upper()
                    exec_row = price_rows.get(symbol) or {}
                    exec_price_raw = _pyfloat(exec_row.get("open"))
                    if exec_price_raw is None:
                        exec_price_raw = _pyfloat(exec_row.get("close"))
                    if exec_price_raw is None and self.ai_broker:
                        exec_price_raw = self.ai_broker.get_latest_price(symbol)
                    # Apply slippage for opening: buying longs pays more, selling shorts gets less
                    if exec_price_raw is not None and ai_slippage_pct > 0:
                        if side == "LONG":
                            exec_price = exec_price_raw * (1 + ai_slippage_pct)
                        else:
                            exec_price = exec_price_raw * (1 - ai_slippage_pct)
                    else:
                        exec_price = exec_price_raw
                    if exec_price is None or exec_price <= 0.0:
                        _record_pipeline_issue(
                            pipeline_stats,
                            "WARNING",
                            f"AI Execution:{symbol}",
                            "Missing execution price for model-driven entry/top-up.",
                        )
                        continue

                    target_dollars = max(0.0, float(trade.get("weight", 0.0) or 0.0) * ai_equity_live)
                    existing = live_lookup.get(symbol)
                    existing_side = str(existing.get("side") or "LONG").upper() if existing is not None else None

                    if existing is not None and existing_side == side:
                        current_market_value = exec_price * float(existing.get("quantity") or 0.0)
                        desired_add = max(0.0, target_dollars - current_market_value)
                        self.ai_tracker.update_position_decision(
                            symbol,
                            trade_date_str,
                            decision_label=trade.get("label"),
                            decision_confidence=trade.get("confidence"),
                            decision_reason=trade.get("reason"),
                            target_price=_pyfloat(existing.get("entry_price")) or exec_price,
                        )
                        if desired_add >= ai_min_trade_dollars and ai_available_capital >= ai_min_trade_dollars:
                            allocation_dollars = min(desired_add, ai_available_capital)
                            quantity = allocation_dollars / exec_price if exec_price else 0.0
                            if quantity > 0.0:
                                added = self.ai_tracker.add_to_position(
                                    symbol=symbol,
                                    add_date=trade_date_str,
                                    add_price=exec_price,
                                    quantity=quantity,
                                    side=side,
                                    target_price=exec_price,
                                    decision_label=trade.get("label"),
                                    decision_confidence=trade.get("confidence"),
                                    decision_reason=trade.get("reason"),
                                    last_decision_date=trade_date_str,
                                )

                                # Execute top-up order on Alpaca if broker is enabled
                                if added and self.ai_broker:
                                    try:
                                        alpaca_side = "buy" if side == "LONG" else "sell"
                                        qty = max(1, int(quantity))
                                        self.ai_broker.submit_order(
                                            symbol=symbol,
                                            qty=qty,
                                            side=alpaca_side,
                                            order_type="market",
                                            time_in_force="day",
                                        )
                                        logger.info(
                                            "Alpaca AI top-up order: %s %s %d shares @ market",
                                            alpaca_side.upper(),
                                            symbol,
                                            qty,
                                        )
                                    except AlpacaError as e:
                                        logger.warning(
                                            "Alpaca AI top-up rejected for %s: %s (code=%s)",
                                            symbol,
                                            e.message,
                                            e.code,
                                        )
                                    except Exception as e:
                                        logger.warning(
                                            "Alpaca AI top-up failed for %s: %s",
                                            symbol,
                                            e,
                                        )

                                if added:
                                    ai_topups.append({
                                        "symbol": symbol,
                                        "side": side,
                                        "entry_price": exec_price,
                                        "target_price": exec_price,
                                        "quantity": quantity,
                                        "allocation_pct": (allocation_dollars / ai_equity_live * 100.0) if ai_equity_live else 0.0,
                                        "allocation_dollars": allocation_dollars,
                                        "reason": f"AI rebalance top-up: {trade.get('reason') or 'target weight increase'}",
                                        "decision_confidence": trade.get("confidence"),
                                        "decision_label": trade.get("label"),
                                    })
                                    ai_available_capital = max(0.0, ai_available_capital - allocation_dollars)
                        continue

                    allocation_dollars = min(target_dollars, ai_available_capital)
                    if allocation_dollars < ai_min_trade_dollars:
                        continue
                    raw_qty = allocation_dollars / exec_price if exec_price else 0.0
                    if raw_qty <= 0.0:
                        continue
                    qty = max(1, int(raw_qty))
                    quantity = float(qty)

                    # Execute order on Alpaca if broker is enabled
                    broker_order = None
                    submit_success = True
                    if self.ai_broker:
                        try:
                            alpaca_side = "buy" if side == "LONG" else "sell"
                            broker_order = self.ai_broker.submit_order(
                                symbol=symbol,
                                qty=qty,
                                side=alpaca_side,
                                order_type="market",
                                time_in_force="day",
                            )
                            logger.info(
                                "Alpaca AI order submitted: %s %s %d shares @ market (order_id=%s)",
                                alpaca_side.upper(),
                                symbol,
                                qty,
                                broker_order.get("id", "unknown"),
                            )
                        except AlpacaError as e:
                            logger.warning(
                                "Alpaca AI order rejected for %s: %s (code=%s). Skipping.",
                                symbol,
                                e.message,
                                e.code,
                            )
                            broker_order = None
                            submit_success = False
                        except Exception as e:
                            logger.warning(
                                "Alpaca AI order failed for %s: %s. Skipping.",
                                symbol,
                                e,
                            )
                            broker_order = None
                            submit_success = False

                    pos_id = None
                    if submit_success:
                        pos_id = self.ai_tracker.open_position(
                            symbol=symbol,
                            entry_date=trade_date_str,
                            entry_price=exec_price,
                            quantity=quantity,
                            side=side,
                            target_price=exec_price,
                            decision_label=trade.get("label"),
                            decision_confidence=trade.get("confidence"),
                            decision_reason=trade.get("reason"),
                            last_decision_date=trade_date_str,
                        )

                    if pos_id:
                        ai_new.append({
                            "symbol": symbol,
                            "side": side,
                            "entry_price": exec_price,
                            "target_price": exec_price,
                            "quantity": quantity,
                            "allocation_pct": (allocation_dollars / ai_equity_live * 100.0) if ai_equity_live else 0.0,
                            "allocation_dollars": allocation_dollars,
                            "reason": trade.get("reason") or "AI target portfolio entry",
                            "decision_confidence": trade.get("confidence"),
                            "decision_label": trade.get("label"),
                        })
                        ai_available_capital = max(0.0, ai_available_capital - allocation_dollars)

            ai_summary = self.ai_tracker.get_portfolio_summary()
            ai_unrealized = self.ai_tracker.get_unrealized_pnl()
            ai_realized_total_dollars = float(ai_summary.get("total_realized_pnl_dollars", 0.0))
            ai_unreal_total_dollars = float(ai_summary.get("total_unrealized_pnl_dollars", 0.0))
            ai_realized_today_dollars = sum([p.get('realized_pnl_dollars', 0.0) for p in ai_closed])
            ai_realized_today = ai_realized_today_dollars / 200000.0

            ai_current_capital = ai_initial_capital + ai_realized_total_dollars + ai_unreal_total_dollars
            ai_invested_notional, ai_available_cash = _position_cash_metrics(ai_unrealized, ai_current_capital)

            if isinstance(ai_llm_status, dict):
                ai_llm_status["target_positions"] = len(ai_trades)
                ai_llm_status["positions_opened"] = len(ai_new)
                ai_llm_status["positions_closed_by_ai"] = len(ai_closed)
                ai_llm_status["positions_topped_up"] = len(ai_topups)

            period_start_ai = self.ai_tracker.get_performance_period_start()
            ai_report = {
                "date": test_date.date(),
                "performance_period_start": period_start_ai,
                "new_positions_opened": len(ai_new),
                "positions_topped_up": len(ai_topups),
                "positions_closed_at_tp": len(ai_closed),
                "positions_closed_by_ai": len(ai_closed),
                "open_positions": int(ai_summary.get("open_positions", 0)),
                "realized_pnl_today": ai_realized_today,
                "realized_pnl_today_dollars": ai_realized_today_dollars,
                "total_realized_pnl": ai_realized_total_dollars / 200000.0,
                "total_realized_pnl_dollars": ai_realized_total_dollars,
                "total_unrealized_pnl": ai_unreal_total_dollars / 200000.0,
                "total_unrealized_pnl_dollars": ai_unreal_total_dollars,
                "total_account_return": (ai_current_capital / 100000.0) / 2.0 - 0.5 + float(ai_previous_return_pct or 0.0),
                "current_capital_estimate": ai_current_capital,
                "invested_notional": ai_invested_notional,
                "available_cash": ai_available_cash,
                "initial_capital": ai_initial_capital,
                "previous_return_pct": float(ai_previous_return_pct or 0.0),
                "ai_position_management_mode": ai_position_management_mode,
                "ai_llm_ok": bool((ai_llm_status or {}).get("ok")) if isinstance(ai_llm_status, dict) else False,
                "ai_llm_error": str((ai_llm_status or {}).get("error") or "") if isinstance(ai_llm_status, dict) else "",
                "ai_llm_skipped_reason": str((ai_llm_status or {}).get("skipped_reason") or "") if isinstance(ai_llm_status, dict) else "",
                "ai_model_used": str((ai_llm_status or {}).get("model_used") or "") if isinstance(ai_llm_status, dict) else "",
                "ai_candidates_seen": int((ai_llm_status or {}).get("candidates_seen") or (ai_llm_status or {}).get("candidates_built") or 0) if isinstance(ai_llm_status, dict) else 0,
                "ai_candidates_scored": int((ai_llm_status or {}).get("candidates_scored") or 0) if isinstance(ai_llm_status, dict) else 0,
            }

            # --- OVERRIDE AI report with Alpaca data when broker is connected ---
            if self.ai_broker:
                try:
                    # Keep a copy of SQLite-based unrealized DataFrame before overriding
                    ai_sqlite_unrealized = ai_unrealized.copy() if not ai_unrealized.empty else pd.DataFrame()

                    ai_alpaca_positions = self.ai_broker.get_positions()
                    ai_alpaca_account = self.ai_broker.get_account()

                    ai_alpaca_equity = float(ai_alpaca_account.get('equity', 0))
                    ai_alpaca_cash = float(ai_alpaca_account.get('cash', 0))

                    # Build Alpaca-sourced unrealized DataFrame
                    ai_alpaca_unrealized_rows = []
                    for p in ai_alpaca_positions:
                        symbol = p.get('symbol', '')
                        side = str(p.get('side', 'long')).upper()
                        qty = float(p.get('qty', 0))
                        entry_price = float(p.get('avg_entry_price', 0))
                        current_price = float(p.get('current_price', 0))
                        unrealized_pl = float(p.get('unrealized_pl', 0))
                        unrealized_plpc = float(p.get('unrealized_plpc', 0))

                        entry_date = None
                        target_price = None
                        if not ai_sqlite_unrealized.empty:
                            match = ai_sqlite_unrealized[ai_sqlite_unrealized['symbol'] == symbol]
                            if not match.empty:
                                entry_date = match.iloc[0].get('entry_date')
                                target_price = match.iloc[0].get('target_price')

                        ai_alpaca_unrealized_rows.append({
                            'symbol': symbol,
                            'side': side,
                            'entry_date': entry_date,
                            'entry_price': entry_price,
                            'quantity': qty,
                            'current_price': current_price,
                            'target_price': target_price,
                            'unrealized_pnl': unrealized_plpc,
                            'unrealized_pnl_dollars': unrealized_pl,
                            'distance_to_tp': abs(target_price - current_price) / current_price if target_price and current_price else None,
                        })

                    # Add any positions from SQLite that are not in Alpaca positions (e.g. newly entered today or historical)
                    ai_alpaca_symbols = {p.get('symbol') for p in ai_alpaca_positions}
                    if not ai_sqlite_unrealized.empty:
                        for _, row in ai_sqlite_unrealized.iterrows():
                            sym = row.get('symbol')
                            if sym and sym not in ai_alpaca_symbols:
                                ai_alpaca_unrealized_rows.append({
                                    'symbol': sym,
                                    'side': row.get('side'),
                                    'entry_date': row.get('entry_date'),
                                    'entry_price': row.get('entry_price'),
                                    'quantity': row.get('quantity'),
                                    'current_price': row.get('current_price'),
                                    'target_price': row.get('target_price'),
                                    'unrealized_pnl': row.get('unrealized_pnl'),
                                    'unrealized_pnl_dollars': row.get('unrealized_pnl_dollars'),
                                    'distance_to_tp': row.get('distance_to_tp'),
                                })

                    ai_unrealized = pd.DataFrame(ai_alpaca_unrealized_rows)

                    # --- Recompute P&L and metrics from the final combined list of open positions ---
                    ai_alpaca_invested = sum(abs(float(row['quantity'] or 0) * float(row['current_price'] or 0)) for row in ai_alpaca_unrealized_rows)
                    ai_alpaca_unrealized = sum(float(row['unrealized_pnl_dollars'] or 0) for row in ai_alpaca_unrealized_rows)
                    ai_alpaca_realized = ai_alpaca_equity - ai_initial_capital - ai_alpaca_unrealized

                    # --- Return calculation ---
                    ai_equity_at_reset = ai_cfg.get('equity_at_reset')
                    if ai_equity_at_reset is None:
                        ai_equity_at_reset = ai_initial_capital + float(ai_previous_return_pct or 0.0) * 200000.0
                    else:
                        ai_equity_at_reset = float(ai_equity_at_reset)
                    ai_new_return = (ai_alpaca_equity - ai_equity_at_reset) / 200000.0
                    ai_total_account_return = (ai_alpaca_equity / 100000.0) / 2.0 - 0.5 + float(ai_previous_return_pct or 0.0)

                    # --- Override report dict ---
                    ai_report['open_positions'] = len(ai_unrealized)
                    ai_report['current_capital_estimate'] = ai_alpaca_equity
                    ai_report['invested_notional'] = ai_alpaca_invested
                    # Calculate available cash as equity minus long positions value to satisfy Accounting Identity
                    ai_alpaca_long_notional = sum(
                        float(row['quantity'] or 0) * float(row['current_price'] or 0)
                        for row in ai_alpaca_unrealized_rows
                        if str(row.get('side', 'LONG')).upper() != 'SHORT'
                    )
                    ai_report['available_cash'] = ai_alpaca_equity - ai_alpaca_long_notional
                    ai_report['total_realized_pnl'] = ai_alpaca_realized / 200000.0
                    ai_report['total_realized_pnl_dollars'] = ai_alpaca_realized
                    ai_report['total_unrealized_pnl'] = ai_alpaca_unrealized / 200000.0
                    ai_report['total_unrealized_pnl_dollars'] = ai_alpaca_unrealized
                    ai_report['total_account_return'] = ai_total_account_return
                    ai_report['previous_return_pct'] = float(ai_previous_return_pct or 0.0)

                    # Override today's trades from Alpaca orders
                    from datetime import date as _date2
                    today_str_ai = str(test_date.date()) if hasattr(test_date, 'date') else str(test_date)
                    ai_all_orders = self.ai_broker.get_orders(status='all', limit=200)

                    ai_seen = set()
                    ai_unique = []
                    for o in ai_all_orders:
                        oid = o.get('id')
                        if oid and oid not in ai_seen:
                            ai_seen.add(oid)
                            ai_unique.append(o)

                    ai_alpaca_new = []
                    ai_alpaca_closed = []
                    ai_realized_today = 0.0

                    for o in ai_unique:
                        created_at = o.get('created_at', '')
                        filled_at = o.get('filled_at', '')
                        is_today = (created_at and str(created_at)[:10] == today_str_ai) or (filled_at and str(filled_at)[:10] == today_str_ai)
                        if not is_today:
                            continue
                        symbol = o.get('symbol', '')
                        side = o.get('side', '')
                        status = o.get('status', '').lower()
                        if status in ('canceled', 'rejected', 'expired'):
                            continue

                        is_filled = (status == 'filled')
                        if not is_filled:
                            continue
                        qty = float(o.get('qty', 0) or 0)
                        qty_abs = abs(qty)
                        filled_avg_price = float(o.get('filled_avg_price', 0) or 0)

                        alp_pos = None
                        for p in ai_alpaca_positions:
                            if p.get('symbol') == symbol:
                                alp_pos = p
                                break

                        is_opening = False
                        is_closing = False
                        if side == 'buy':
                            if alp_pos and str(alp_pos.get('side', '')).lower() == 'short':
                                is_closing = True
                            else:
                                is_opening = True
                        elif side == 'sell':
                            if alp_pos and str(alp_pos.get('side', '')).lower() == 'long':
                                is_closing = True
                            else:
                                is_opening = True

                        local_match = None
                        for p in ai_new:
                            if p.get('symbol') == symbol:
                                local_match = p
                                break

                        if is_opening:
                            target_price = None
                            if local_match:
                                target_price = local_match.get('target_price')
                                notional = local_match.get('allocation_dollars', 0.0)
                                alloc_pct = local_match.get('allocation_pct', 0.0)
                                entry_p = filled_avg_price if is_filled else local_match.get('entry_price')
                                qty_abs = local_match.get('quantity', qty_abs)
                                reason = local_match.get('reason')
                                conf = local_match.get('decision_confidence')
                                label = local_match.get('decision_label')
                            else:
                                if not ai_unrealized.empty:
                                    m = ai_unrealized[ai_unrealized['symbol'] == symbol]
                                    if not m.empty:
                                        target_price = m.iloc[0].get('target_price')
                                entry_p = filled_avg_price if is_filled else 0.0
                                notional = entry_p * qty_abs
                                alloc_pct = (notional / ai_alpaca_equity * 100) if ai_alpaca_equity else 0.0
                                reason = "AI target portfolio entry"
                                conf = None
                                label = None

                            side_label = 'LONG' if side == 'buy' else 'SHORT'
                            if not is_filled:
                                side_label += ' (Pending)'

                            ai_alpaca_new.append({
                                'symbol': symbol,
                                'side': side_label,
                                'entry_price': entry_p if is_filled else "Pending",
                                'target_price': target_price,
                                'allocation_pct': alloc_pct,
                                'allocation_dollars': notional,
                                'quantity': qty_abs,
                                'reason': reason,
                                'decision_confidence': conf,
                                'decision_label': label,
                            })
                        elif is_closing:
                            r_dollars = 0.0
                            r_pct = 0.0
                            entry_p = 0.0
                            exit_p = filled_avg_price if is_filled else 0.0

                            local_close_match = None
                            for p in ai_closed:
                                if p.get('symbol') == symbol:
                                    local_close_match = p
                                    break

                            if alp_pos:
                                entry_p = float(alp_pos.get('avg_entry_price', 0))
                                ps = str(alp_pos.get('side', '')).lower()
                                if is_filled:
                                    if ps == 'long':
                                        r_dollars = (exit_p - entry_p) * qty_abs
                                    else:
                                        r_dollars = (entry_p - exit_p) * qty_abs
                                    r_pct = r_dollars / (entry_p * qty_abs) if entry_p and qty_abs else 0.0
                            elif local_close_match:
                                entry_p = local_close_match.get('entry_price', 0.0)
                                if is_filled:
                                    r_dollars = local_close_match.get('realized_pnl_dollars', 0.0)
                                    r_pct = local_close_match.get('realized_pnl', 0.0)

                            ai_realized_today += r_dollars
                            side_label = 'LONG' if side == 'sell' else 'SHORT'
                            if not is_filled:
                                side_label += ' (Pending Exit)'

                            ai_alpaca_closed.append({
                                'symbol': symbol,
                                'side': side_label,
                                'entry_price': entry_p,
                                'exit_price': exit_p if is_filled else "Pending",
                                'realized_pnl': r_pct if is_filled else 0.0,
                                'realized_pnl_dollars': r_dollars if is_filled else 0.0,
                                'reason': local_close_match.get('reason') if local_close_match else "AI rotation exit",
                            })

                    ai_new = ai_alpaca_new
                    ai_closed = ai_alpaca_closed
                    ai_report['positions_closed_at_tp'] = len(ai_alpaca_closed)
                    ai_report['positions_closed_by_ai'] = len(ai_alpaca_closed)
                    ai_report['new_positions_opened'] = len(ai_alpaca_new)
                    ai_report['realized_pnl_today_dollars'] = ai_realized_today
                    ai_report['realized_pnl_today'] = ai_realized_today / 200000.0

                    logger.info(
                        "Alpaca AI FULL override: equity=$%.2f, realized=$%.2f, unrealized=$%.2f, return=%.2f%%",
                        ai_alpaca_equity, ai_alpaca_realized, ai_alpaca_unrealized, ai_total_account_return * 100,
                    )
                except Exception as e:
                    logger.warning("Failed to pull Alpaca AI data for report: %s", e)
                    import traceback
                    logger.warning(traceback.format_exc())

            pd.DataFrame([ai_report]).to_csv(
                os.path.join(self.results_dir, f"daily_report_ai_{date_str}.csv"),
                index=False
            )
            if not ai_unrealized.empty:
                ai_unrealized.to_csv(
                    os.path.join(self.results_dir, f"unrealized_ai_{date_str}.csv"),
                    index=False
                )

        if pipeline_stats is None:
            pipeline_stats = {}
        if isinstance(pipeline_stats, dict):
            if ai_enabled and isinstance(ai_llm_status, dict) and not bool(ai_llm_status.get("ok")):
                _record_pipeline_issue(
                    pipeline_stats,
                    "ERROR",
                    "AI Trading",
                    ai_llm_status.get("error") or "AI trading decision call failed.",
                )
            _finalize_pipeline_health(pipeline_stats)

        # Keep Core and AI reporting stats separate to avoid showing AI-trading fields
        # in the Core email.
        core_pipeline_stats = dict(pipeline_stats) if isinstance(pipeline_stats, dict) else pipeline_stats
        ai_pipeline_stats = dict(pipeline_stats) if isinstance(pipeline_stats, dict) else pipeline_stats
        if isinstance(core_pipeline_stats, dict):
            core_pipeline_stats.pop("ai_trading_llm_status", None)
        if isinstance(ai_pipeline_stats, dict):
            ai_pipeline_stats["ai_trading_llm_status"] = ai_llm_status

        # Send TWO separate emails (Core + AI)
        from email_notifier import EmailNotifier
        notifier = EmailNotifier()
        core_email_sent = True
        if report is not None:
            core_email_sent = notifier.send_daily_report(
                report_data=report,
                unrealized_df=unrealized,
                closed_positions=closed_positions,
                new_positions=entries_today,
                meta_insights=None,
                signal_rankings=core_rank_df,
                pipeline_stats=core_pipeline_stats,
                backtest_signals=backtest_signals,
                subject_tag="Core"
            )

        ai_email_sent = True
        if ai_report is not None:
            backend_provider = str(ai_llm_status.get('backend_provider') or '').strip()
            selected_backend = str(ai_llm_status.get('selected_backend') or ai_llm_status.get('backend') or '').strip()
            model_used = str(ai_llm_status.get('model_used') or ai_llm_status.get('model') or '').strip()
            fallback_from_backend = str(ai_llm_status.get('fallback_from_backend') or '').strip()
            backend_summary = selected_backend or backend_provider or "unknown"
            if backend_provider and backend_provider.lower() not in backend_summary.lower():
                backend_summary = f"{backend_summary} ({backend_provider})"
            if ai_llm_status.get("ok"):
                ai_insight = (
                    "AI trading engine status: OK"
                    f" | mode={ai_llm_status.get('manager_mode') or 'unknown'}"
                    f" | backend={backend_summary}"
                    f" | model={model_used or 'unknown'}"
                    f" | target_positions={ai_llm_status.get('target_positions', 0)}"
                    f" | closed={ai_llm_status.get('positions_closed_by_ai', 0)}"
                    f" | opened={ai_llm_status.get('positions_opened', 0)}"
                    f" | topped_up={ai_llm_status.get('positions_topped_up', 0)}"
                )
                if fallback_from_backend:
                    ai_insight += f" | fallback_from={fallback_from_backend}"
            else:
                ai_insight = (
                    f"AI trading engine status: ERROR - {ai_llm_status.get('error')}"
                    f" | backend={backend_summary}"
                    f" | model={model_used or 'unknown'}"
                )
            ai_email_positions = list(ai_new) + list(ai_topups)
            ai_email_sent = notifier.send_daily_report(
                report_data=ai_report,
                unrealized_df=ai_unrealized,
                closed_positions=ai_closed,
                new_positions=ai_email_positions,
                meta_insights=ai_insight,
                signal_rankings=None,
                pipeline_stats=ai_pipeline_stats,
                backtest_signals=backtest_signals,
                subject_tag="AI"
            )

        # In AI-enabled runs, both strategy paths must complete cleanly.
        # Core-only maintenance runs can still succeed without an AI email.
        if ai_enabled:
            if report is None:
                email_sent = bool(ai_email_sent) and (ai_report is not None)
            else:
                email_sent = bool(core_email_sent) and bool(ai_email_sent) and (ai_report is not None)
        else:
            email_sent = bool(core_email_sent) if report is not None else True
        if email_sent:
            markers = []
            if report is not None:
                markers.append(f"email_sent_core_{date_str}.ok")
            if ai_report is not None:
                markers.append(f"email_sent_ai_{date_str}.ok")
            if report is not None and ai_report is not None:
                markers.append(f"email_sent_{date_str}.ok")
            for marker in markers:
                marker_path = os.path.join(self.results_dir, marker)
                try:
                    with open(marker_path, "w") as handle:
                        handle.write(get_sgt_now().isoformat())
                except Exception as exc:
                    logger.warning(f"Failed to write email sent marker: {exc}")
        else:
            logger.warning("One or more required strategy emails were not sent; scheduler will allow retry.")

        # Cleanup: Keep only last 2 days of files
        self._cleanup_old_files(self.results_dir, keep_days=2)

        return report, unrealized, closed_positions, email_sent

    def _cleanup_old_files(self, results_dir, keep_days=2):
        """Delete result files older than keep_days"""
        now = datetime.now()
        for file in os.listdir(results_dir):
            if not (file.endswith('.csv') or file.endswith('.log')):
                continue
            path = os.path.join(results_dir, file)
            mtime = datetime.fromtimestamp(os.path.getmtime(path))
            if (now - mtime).days > keep_days:
                try:
                    os.remove(path)
                    logger.info(f"Deleted old file: {file}")
                except Exception as exc:
                    logger.warning(f"Failed to delete {file}: {exc}")


def run_daily_job(config_path=None):
    """
    Fast daily job for large universes (S&P 500):
    - Update latest prices (API if available, otherwise Stooq fallback)
    - Fetch news/LLM sentiment only for currently-held symbols (keeps runtime bounded)
    - Run daily position-tracking backtest and send emails
    - Apply storage policy (retention/vacuum/prune)
    """
    config_path = _get_config_path(config_path)
    config = _load_config(config_path)
    base_dir = os.path.dirname(os.path.abspath(config_path))
    universe_path = _resolve_path(base_dir, config['universe']['source'])

    # Check if database reset is requested via environment variable
    if os.getenv("RESET_CORE_DATABASE", "").strip().lower() in {"1", "true", "yes", "on"}:
        logger.info("RESET_CORE_DATABASE environment variable is set. Resetting database state before price ingestion...")
        try:
            from reset_performance import reset_all_state
            reset_all_state(strategy="core")
            logger.info("Reset complete. Starting clean run.")
        except Exception as e:
            logger.error("Failed to execute database reset: %s", e)

    universe_df = pd.read_csv(universe_path)
    tickers = [str(t).strip().upper() for t in universe_df['ticker'].tolist() if str(t).strip()]
    ai_cfg = config.get("ai_trading", {}) if isinstance(config, dict) else {}
    ai_disabled_by_env = str(os.getenv("DISABLE_AI_TRADING", "")).strip().lower() in {
        "1",
        "true",
        "yes",
        "on",
    }
    ai_enabled = bool(ai_cfg.get("enabled", False)) and not ai_disabled_by_env
    core_disabled_by_env = str(os.getenv("DISABLE_CORE_TRADING", "")).strip().lower() in {
        "1",
        "true",
        "yes",
        "on",
    }
    skip_core_model_training = bool(core_disabled_by_env and ai_enabled)

    price_ingestor = PriceIngestor(config_path)
    providers = (config.get("data", {}).get("providers", {}) or {})
    td = providers.get("twelvedata", {}) if isinstance(providers, dict) else {}
    assumed_rpm = int(td.get("assumed_requests_per_minute_per_key", 5) or 5)
    key_count = len(price_ingestor.twelvedata_keys.keys())
    # Throttle globally so per-key usage stays under free-tier limits.
    # overall_rps ~= keys * rpm / 60
    overall_rps = (key_count * assumed_rpm) / 60.0 if key_count > 0 else 0.0
    sleep_s = 1.0 / overall_rps if overall_rps > 0 else 0.6
    sleep_s = max(0.2, min(2.0, sleep_s))

    latest_market_date = None
    try:
        latest_market_date = price_ingestor.get_latest_market_date()
    except Exception:
        latest_market_date = None

    results_dir = _resolve_path(
        base_dir,
        (config.get("output", {}) or {}).get("results_dir", "./results")
    )

    # Determine the provider's latest available market date using a single "sentinel" symbol.
    # This avoids a deadlock where the scheduler won't run because DB market date is stale.
    src = str(getattr(price_ingestor, "price_source", "stooq") or "stooq").strip().lower()
    if src == "auto":
        if price_ingestor.twelvedata_keys.keys():
            src = "twelvedata"
        elif price_ingestor.alphavantage_keys.keys():
            src = "alphavantage"
        else:
            src = "stooq"

    pipeline_stats = {
        'start_time': get_sgt_now().isoformat(),
        'steps': [],
        'tickers_total': len(tickers),
        'tickers_processed': 0,
        'tickers_failed': 0,
        'news_fetched': 0,
        'news_enabled': bool(config.get('news', {}).get('enabled', False)),
        'llm_status': None,
        'error_count': 0,
        'warning_count': 0,
        'issues': [],
        'issue_overflow_count': 0,
        'failed_symbols': [],
    }
    
    def log_step(name, status, details=None):
        pipeline_stats['steps'].append({
            'time': get_sgt_now().strftime("%H:%M:%S"),
            'step': name,
            'status': status,
            'details': details
        })
        logger.info(f"STEP: {name} | STATUS: {status} | {details or ''}")
        status_u = str(status or "").strip().upper()
        if status_u == "FAILED":
            _record_pipeline_issue(pipeline_stats, "ERROR", name, details or "step failed")
        elif status_u == "WARNING":
            _record_pipeline_issue(pipeline_stats, "WARNING", name, details or "step warning")

    # One-time recovery for cloud cache misses: seed only if tables are empty.
    log_step("State Recovery", "Started")
    try:
        recovery = recover_positions_from_seed(config_path)
        if recovery.get("recovered_total", 0) > 0:
            log_step(
                "State Recovery",
                "Completed",
                f"Recovered core={recovery.get('recovered_core', 0)} ai={recovery.get('recovered_ai', 0)}",
            )
        else:
            log_step("State Recovery", "Skipped", recovery.get("reason", "no_recovery_needed"))
    except Exception as exc:
        log_step("State Recovery", "Failed", str(exc))

    # Safety cleanup: remove stale OPEN rows that match static seed entries.
    log_step("Seed Sanity", "Started")
    try:
        purge = purge_seeded_open_positions(config_path)
        purged_total = int(purge.get("purged_total", 0) or 0)
        if purged_total > 0:
            log_step(
                "Seed Sanity",
                "Completed",
                f"Purged core={purge.get('purged_core', 0)} ai={purge.get('purged_ai', 0)}",
            )
        else:
            log_step("Seed Sanity", "Skipped", purge.get("reason", "none_matched"))
    except Exception as exc:
        log_step("Seed Sanity", "Failed", str(exc))

    # Safety pass: if historical open positions exceed account capital,
    # shrink quantities before today's strategy logic runs.
    log_step("Position Sanity", "Started")
    try:
        sanity = enforce_position_cap(config_path)
        adjusted_total = int(sanity.get("adjusted_total", 0) or 0)
        if adjusted_total > 0:
            details = []
            core_tbl = (sanity.get("tables", {}) or {}).get("positions", {}) or {}
            ai_tbl = (sanity.get("tables", {}) or {}).get("positions_ai", {}) or {}
            if core_tbl.get("adjusted"):
                details.append(f"core={core_tbl.get('adjusted')}")
            if ai_tbl.get("adjusted"):
                details.append(f"ai={ai_tbl.get('adjusted')}")
            log_step("Position Sanity", "Completed", "Adjusted " + ", ".join(details))
        else:
            log_step("Position Sanity", "Skipped", "within_cap")
    except Exception as exc:
        log_step("Position Sanity", "Failed", str(exc))

    log_step("Market Check", "Started", f"Source: {src}")

    sentinel = "AAPL" if "AAPL" in tickers else (tickers[0] if tickers else "AAPL")
    provider_latest_date = None
    try:
        if src == "twelvedata":
            df = price_ingestor.fetch_twelvedata_daily(sentinel, outputsize=5)
            if df is not None and not df.empty and "date" in df.columns:
                provider_latest_date = str(df["date"].max())
        elif src == "alphavantage":
            df = price_ingestor.fetch_alphavantage_daily(sentinel, outputsize="compact")
            if df is not None and not df.empty and "date" in df.columns:
                provider_latest_date = str(df["date"].max())
        else:
            df = price_ingestor.fetch_stooq_latest(sentinel)
            if df is not None and not df.empty and "date" in df.columns:
                provider_latest_date = str(df["date"].max())
    except Exception:
        provider_latest_date = None

    # If no new provider data is available AND this strategy leg has already
    # emailed for the latest DB market date, do nothing.  The scheduled GitHub
    # workflow runs Core and AI as separate invocations, so the Core email
    # marker must never suppress the later AI-only email.
    from datetime import datetime
    import pytz
    est_tz = pytz.timezone('America/New_York')
    today_est = datetime.now(est_tz).strftime('%Y-%m-%d')
    is_today = (provider_latest_date == today_est)

    if latest_market_date and provider_latest_date and provider_latest_date <= latest_market_date and not is_today:
        date_token = str(latest_market_date).replace('-', '')
        if core_disabled_by_env and ai_enabled:
            marker_names = [f"email_sent_ai_{date_token}.ok"]
        elif ai_disabled_by_env or not ai_enabled:
            marker_names = [f"email_sent_core_{date_token}.ok"]
        else:
            marker_names = [f"email_sent_{date_token}.ok"]
        existing_marker = next(
            (
                os.path.join(results_dir, marker_name)
                for marker_name in marker_names
                if os.path.exists(os.path.join(results_dir, marker_name))
            ),
            None,
        )
        if existing_marker:
            logger.info(
                "No new market data (provider_latest=%s, db_latest=%s) and %s already exists. Skipping run.",
                provider_latest_date,
                latest_market_date,
                os.path.basename(existing_marker),
            )
            return True

    open_position_symbols = _get_open_position_symbols(config_path)
    ticker_set = set(tickers)
    held_extra_symbols = [symbol for symbol in open_position_symbols if symbol not in ticker_set]
    pipeline_stats["held_symbols_refreshed"] = 0
    pipeline_stats["held_symbols_refresh_failed"] = 0

    resolved_price_source = str(getattr(price_ingestor, "price_source", "stooq") or "stooq").strip().lower()
    if resolved_price_source == "auto":
        if price_ingestor.twelvedata_keys.keys():
            resolved_price_source = "twelvedata"
        elif price_ingestor.alphavantage_keys.keys():
            resolved_price_source = "alphavantage"
        else:
            resolved_price_source = "stooq"

    shared_broker = None

    def _refresh_symbol_prices(symbol):
        sym_last = None
        try:
            sym_last = price_ingestor.get_latest_date_for_symbol(symbol)
        except Exception:
            sym_last = None

        df_prices = None
        if sym_last is None:
            # First time: limited window.
            if resolved_price_source == "twelvedata":
                df_prices = price_ingestor.fetch_twelvedata_daily(symbol)
                if df_prices is None or df_prices.empty:
                    if price_ingestor.alphavantage_keys.keys():
                        df_prices = price_ingestor.fetch_alphavantage_daily(symbol, outputsize="compact")
                    if df_prices is None or df_prices.empty:
                        df_prices = price_ingestor.fetch_stooq_data(symbol)
            elif resolved_price_source == "alphavantage":
                df_prices = price_ingestor.fetch_alphavantage_daily(symbol, outputsize="compact")
                if df_prices is None or df_prices.empty:
                    df_prices = price_ingestor.fetch_stooq_data(symbol)
            else:
                df_prices = price_ingestor.fetch_stooq_data(symbol)
        else:
            # Incremental
            from datetime import datetime
            import pytz
            est_tz = pytz.timezone('America/New_York')
            today_est = datetime.now(est_tz).strftime('%Y-%m-%d')
            is_today = (provider_latest_date == today_est)

            if provider_latest_date and sym_last == provider_latest_date and not is_today:
                df_prices = None
            elif resolved_price_source == "twelvedata":
                df_prices = price_ingestor.fetch_twelvedata_daily(symbol, outputsize=5)
                if (df_prices is None or df_prices.empty) and price_ingestor.alphavantage_keys.keys():
                    df_prices = price_ingestor.fetch_alphavantage_daily(symbol, outputsize="compact")
                if df_prices is None or df_prices.empty:
                    df_prices = price_ingestor.fetch_stooq_latest(symbol)
            elif resolved_price_source == "alphavantage":
                df_prices = price_ingestor.fetch_alphavantage_daily(symbol, outputsize="compact")
                if df_prices is None or df_prices.empty:
                    df_prices = price_ingestor.fetch_stooq_latest(symbol)
            else:
                df_prices = price_ingestor.fetch_stooq_latest(symbol)

        if df_prices is None or df_prices.empty:
            nonlocal shared_broker
            try:
                if shared_broker is None:
                    shared_broker = create_broker_from_env("core") or create_broker_from_env("ai")
            except Exception:
                shared_broker = None
            if shared_broker is not None:
                try:
                    bars = shared_broker.get_bars(symbol, timeframe="1Day", limit=5)
                    if bars:
                        rows = []
                        for bar in bars:
                            dt_str = bar.get("t", "")[:10]
                            if dt_str:
                                rows.append({
                                    "symbol": symbol,
                                    "date": dt_str,
                                    "open": float(bar.get("o", 0.0) or 0.0),
                                    "high": float(bar.get("h", 0.0) or 0.0),
                                    "low": float(bar.get("l", 0.0) or 0.0),
                                    "close": float(bar.get("c", 0.0) or 0.0),
                                    "volume": int(float(bar.get("v", 0.0) or 0.0)),
                                })
                        if rows:
                            df_prices = pd.DataFrame(rows)
                except Exception as e:
                    logger.warning("Alpaca broker price ingestion fallback failed for %s: %s", symbol, e)

        if df_prices is not None and not df_prices.empty:
            conn = sqlite3.connect(price_ingestor.db_path)
            try:
                price_ingestor._sqlite_upsert(
                    type('Table', (), {'name': 'prices'}),
                    conn,
                    df_prices.columns.tolist(),
                    df_prices.values.tolist()
                )
            finally:
                conn.close()

    # Update prices for scan-universe tickers (incremental) without downloading full history repeatedly.
    log_step("Price Ingestion", "Started", f"Processing {len(tickers)} symbols...")
    if skip_core_model_training:
        logger.info(
            "Core trading disabled; skipping per-ticker OLS retraining during AI-only daily job."
        )
    for t in tickers:
        try:
            _refresh_symbol_prices(t)
            pipeline_stats['tickers_processed'] += 1

            time.sleep(sleep_s)
        except Exception as exc:
            pipeline_stats['tickers_failed'] += 1
            logger.warning(f"Failed to ingest {t}: {exc}")
            if len(pipeline_stats.get("failed_symbols", [])) < 25:
                pipeline_stats["failed_symbols"].append(str(t))
            _record_pipeline_issue(pipeline_stats, "ERROR", f"Price Ingestion:{t}", str(exc))
            continue

    log_step("Price Ingestion", "Completed", f"Success: {pipeline_stats['tickers_processed']}, Failed: {pipeline_stats['tickers_failed']}")

    # Always refresh open positions too, even if they were opened from an older/larger universe.
    # Otherwise the report can silently show stale cached prices for held names that are no longer
    # in today's scan universe.
    if held_extra_symbols:
        log_step(
            "Held Position Price Refresh",
            "Started",
            f"Refreshing {len(held_extra_symbols)} off-universe held symbols...",
        )
        for symbol in held_extra_symbols:
            try:
                _refresh_symbol_prices(symbol)
                pipeline_stats["held_symbols_refreshed"] = int(pipeline_stats.get("held_symbols_refreshed", 0) or 0) + 1
                time.sleep(sleep_s)
            except Exception as exc:
                pipeline_stats["held_symbols_refresh_failed"] = int(pipeline_stats.get("held_symbols_refresh_failed", 0) or 0) + 1
                _record_pipeline_issue(pipeline_stats, "ERROR", f"Held Position Price Refresh:{symbol}", str(exc))
                continue
        log_step(
            "Held Position Price Refresh",
            "Completed",
            f"Success: {pipeline_stats['held_symbols_refreshed']}, Failed: {pipeline_stats['held_symbols_refresh_failed']}",
        )
    else:
        log_step("Held Position Price Refresh", "Skipped", "No off-universe held symbols.")

    expected_quote_date = str(provider_latest_date or "") if provider_latest_date else None
    if open_position_symbols and expected_quote_date:
        stale_open_quotes = []
        from datetime import datetime
        expected_dt = None
        try:
            expected_dt = datetime.strptime(expected_quote_date, "%Y-%m-%d")
        except Exception:
            expected_dt = None

        active_tables = []
        if not core_disabled_by_env:
            active_tables.append("positions")
        if ai_enabled:
            active_tables.append("positions_ai")
        if not active_tables:
            active_tables = ["positions"]

        for symbol in open_position_symbols:
            sym_latest = None
            try:
                sym_latest = price_ingestor.get_latest_date_for_symbol(symbol)
            except Exception:
                sym_latest = None

            is_stale = False
            if not sym_latest:
                is_stale = True
            else:
                try:
                    latest_dt = datetime.strptime(str(sym_latest), "%Y-%m-%d")
                    if expected_dt and (expected_dt - latest_dt).days > 10:
                        logger.warning("Position %s has been stale for %d days. Closing in database.", symbol, (expected_dt - latest_dt).days)
                        import sqlite3 as sql3
                        for tbl in active_tables:
                            try:
                                with sqlite3.connect(price_ingestor.db_path) as conn:
                                    row = conn.execute(
                                        f"SELECT id, entry_price FROM {tbl} WHERE symbol=? AND status='OPEN'", (symbol,)
                                    ).fetchone()
                                    if row:
                                        conn.execute(
                                            f"UPDATE {tbl} SET status='CLOSED', exit_date=?, exit_price=?, realized_pnl=0.0, exit_reason='stale_quote_purge' WHERE id=?",
                                            (sym_latest, row[1], row[0])
                                        )
                                        conn.commit()
                            except sql3.OperationalError:
                                pass
                        continue
                    elif expected_dt and latest_dt < expected_dt:
                        is_stale = True
                except Exception:
                    is_stale = True

            if is_stale:
                stale_open_quotes.append(f"{symbol}:{sym_latest or 'missing'}")
        if stale_open_quotes:
            details = (
                f"Open-position quotes are stale vs {expected_quote_date}: "
                + ", ".join(stale_open_quotes[:10])
            )
            log_step("Held Position Quote Freshness", "Warning", details)
        else:
            log_step(
                "Held Position Quote Freshness",
                "Completed",
                f"All open-position quotes current through {expected_quote_date}",
            )
    elif open_position_symbols:
        log_step("Held Position Quote Freshness", "Skipped", "Provider latest market date unavailable.")
    else:
        log_step("Held Position Quote Freshness", "Skipped", "No open positions.")

    # News/LLM sentiment only for held symbols (keeps runtime within budget).
    news_enabled = pipeline_stats['news_enabled']
    if news_enabled:
        log_step("News Ingestion", "Started")
        try:
            news_ingestor = NewsIngestor(config_path)
            core_tracker = PositionTracker(config_path, table_name="positions")
            ai_tracker = PositionTracker(config_path, table_name="positions_ai")
            held = set()
            for df in [core_tracker.get_open_positions(), ai_tracker.get_open_positions()]:
                if df is not None and hasattr(df, "empty") and not df.empty:
                    held |= set([str(s).strip().upper() for s in df["symbol"].tolist() if str(s).strip()])
            
            for sym in sorted(list(held))[:50]:
                try:
                    news_ingestor.fetch_and_store_news(sym)
                    pipeline_stats['news_fetched'] += 1
                except Exception as exc:
                    _record_pipeline_issue(pipeline_stats, "WARNING", f"News Ingestion:{sym}", str(exc))
                    continue
            try:
                pipeline_stats['llm_status'] = news_ingestor.get_llm_status()
                llm_status = pipeline_stats.get('llm_status') if isinstance(pipeline_stats, dict) else None
                if isinstance(llm_status, dict) and int(llm_status.get("errors", 0) or 0) > 0:
                    _record_pipeline_issue(
                        pipeline_stats,
                        "WARNING",
                        "News Sentiment",
                        llm_status.get("last_error") or f"errors={llm_status.get('errors')}",
                    )
            except Exception as exc:
                _record_pipeline_issue(pipeline_stats, "WARNING", "News Sentiment", str(exc))
            log_step("News Ingestion", "Completed", f"Fetched for {pipeline_stats['news_fetched']} symbols")
        except Exception as exc:
            log_step("News Ingestion", "Failed", str(exc))
    else:
        log_step("News Ingestion", "Skipped", "Disabled in config")

    # Run daily backtest + emails
    log_step("Backtest & Strategy", "Started")
    backtester = DailyBacktester(config_path)
    _finalize_pipeline_health(pipeline_stats)
    result = backtester.run_daily_test(pipeline_stats=pipeline_stats)
    email_sent = bool(result and result[-1])
    allow_missing_email = str(os.getenv("ALLOW_MISSING_EMAIL", "")).strip().lower() in {"1", "true", "yes", "on"}

    if email_sent:
        log_step("Backtest & Strategy", "Completed", "Report sent")
    else:
        log_step("Backtest & Strategy", "Warning", "Email failed or no signals")
        if allow_missing_email:
            log_step("Backtest & Strategy", "Completed", "Continuing despite missing email (ALLOW_MISSING_EMAIL=1)")
            email_sent = True

    # Enforce storage limits
    try:
        from storage_policy import apply_storage_policy
        apply_storage_policy(config_path)
    except Exception:
        pass

    return email_sent


if __name__ == "__main__":
    import sys
    success = run_daily_job()
    sys.exit(0 if success else 1)