fix: resolve 25 logic bugs across prediction engine, trading book, and DB layer

HIGH:
- ai_forecast: fix off-by-one in positional arg parsing (ml/nifty_ok/macro_ok/vix/news
were all reading wrong args[N] indices — every prediction ran with ml={}, vix=15.0 fixed)
- predictor_core: _is_actionable_buy checked "BUY"/"STRONG BUY" (never produced);
fix to "BULLISH"/"SLIGHTLY BULLISH" so entry-price buffer actually fires
- macro_context + predictor_core: macro gate always returned Risk-ON due to shift(1) →
today's date missing from index → KeyError swallowed → default True; fix via
build_mask(ffill) + drop NaN warmup rows (bool(NaN)=True was silently passing gate)
- database: close_trade UPDATE had no AND status='OPEN' guard → double-close race
corrupted pnl and duplicated signal_accuracy rows; add guard + rowcount==1 check
- app: _trade_price_diagnostics called fetch_ohlcv with 3 args (takes 2) → silent
TypeError → every postmortem had empty diagnostics; fix to period= kwarg
- database: partial close P&L was computed but never stored; add realized_pnl column
with migration and accumulate on each partial close
- app: auto-close with unavailable live price used entry_price as exit → 0% P&L stored
permanently; now skips close and retries next cycle with WARNING log

MEDIUM:
- ai_forecast: JSON regex {[^{}]*} fails when "reasoning" contains literal braces;
replace with balanced-brace scanner
- app: BEARISH validation actual_return sign-flipped so positive=fell-as-predicted
- app: duplicate _safe_float at line 1708 shadowed the canonical one; removed duplicate
and fixed 3 callers that passed a default= arg
- database: fill_order never updated entry_price on fill; now persists actual fill price
- app: limit order had double 0.1% tolerance dead band (creation + fill check);
fill check now uses live <= limit with no extra margin
- app: _TOP5_COMPUTING check-and-set was non-atomic; protect with threading.Lock
- data_sources: YF 5d daily fallback lacked IST freshness check; could return
yesterday's close on weekends/holidays as "live"

LOW:
- fred_data: >= 12 guard before iloc[-13] → IndexError on 12-item series; fix to >= 13
- database: snapshot dedup used UTC date(); switch to IST (+5:30) offset
- predictor_core: Nifty gate trivially passed with < 3 bars; require len >= 3
- predictor_core: earnings_dates iteration in undocumented order; sort by abs days_away
- predictor_core: new_win stat uncapped; corrupt DB row could inflate all EVs; cap at 20x
- predictor_core: _warn_nifty_unavailable_once check-and-set not thread-safe; add Lock
- predictor_core: add threading import (was missing)
- macro_context: print() → logging.warning() for macro download failures
- app: shares not validated > 0 on trade open; add explicit check
- app: watchlist cache not evicted on ticker delete; add _WATCHLIST_PICK_CACHE.pop()
- app: closing PENDING trade returned 200 OK; now returns 409 Conflict
- database: merge_into_position read+write in same connection to prevent share loss
- trial_run: ADX division by ATR=0 for frozen stocks → inf/NaN; guard with .replace(0, NaN)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

ai_forecast.py

@@ -699,22 +699,42 @@ def _parse_json_from_llm(text: str) -> Dict | None:
     except Exception:
         pass
     if parsed is None:
-        # Try first {...} block — handles "thinking aloud" models that prepend reasoning text
-        m = re.search(r"\{[^{}]*\}", cleaned, re.DOTALL)
-        if m:
-            try:
-                parsed = json.loads(m.group(0))
-            except Exception:
-                pass
-    if parsed is None:
-        # Broader search: find the last {...} block (some models put JSON at the end)
-        matches = list(re.finditer(r"\{[^{}]*\}", cleaned, re.DOTALL))
-        for candidate in reversed(matches):
-            try:
-                parsed = json.loads(candidate.group(0))
-                break
-            except Exception:
-                pass
+        # Balanced-brace scan: find the outermost {...} block.
+        # {[^{}]*} only matches flat objects and breaks when "reasoning" contains
+        # literal braces like "RSI crossed {35}". Walk the string instead.
+        def _find_json_object(s: str):
+            for start in range(len(s)):
+                if s[start] != '{':
+                    continue
+                depth, in_str, i = 0, False, start
+                while i < len(s):
+                    ch = s[i]
+                    if ch == '"' and (i == 0 or s[i - 1] != '\\'):
+                        in_str = not in_str
+                    if not in_str:
+                        if ch == '{':
+                            depth += 1
+                        elif ch == '}':
+                            depth -= 1
+                            if depth == 0:
+                                candidate = s[start:i + 1]
+                                try:
+                                    return json.loads(candidate)
+                                except Exception:
+                                    break
+                    i += 1
+            return None
+
+        parsed = _find_json_object(cleaned)
+        if parsed is None:
+            # Last-resort: try scanning from the end for the last JSON object
+            for start in range(len(cleaned) - 1, -1, -1):
+                if cleaned[start] == '{':
+                    try:
+                        parsed = json.loads(cleaned[start:])
+                        break
+                    except Exception:
+                        pass
     if parsed is None:
         return None
     # Reject if any required forecast field is missing — prevents 0.0 default corruption
@@ -1236,11 +1256,11 @@ def get_ai_forecast(
     _TF_LABELS = ("INTRADAY", "1D", "3D", "5D")
     company       = args[0] if len(args) >= 1 and isinstance(args[0], str) and args[0] not in _TF_LABELS else kwargs.get("company", ticker)
     tf_label      = next((a for a in args if a in _TF_LABELS), None) or kwargs.get("tf_label", "3D")
-    ml            = args[3] if len(args) >= 4 and isinstance(args[3], dict) else kwargs.get("ml", {})
-    nifty_ok      = bool(args[4]) if len(args) >= 5 else bool(kwargs.get("nifty_ok", True))
-    macro_ok      = bool(args[5]) if len(args) >= 6 else bool(kwargs.get("macro_ok", True))
-    vix_level     = float(args[6]) if len(args) >= 7 and isinstance(args[6], (int,float)) else float(kwargs.get("vix_level", 15.0))
-    news          = args[7] if len(args) >= 8 and isinstance(args[7], dict) else kwargs.get("news", {})
+    ml            = args[2] if len(args) >= 3 and isinstance(args[2], dict) else kwargs.get("ml", {})
+    nifty_ok      = bool(args[3]) if len(args) >= 4 else bool(kwargs.get("nifty_ok", True))
+    macro_ok      = bool(args[4]) if len(args) >= 5 else bool(kwargs.get("macro_ok", True))
+    vix_level     = float(args[5]) if len(args) >= 6 and isinstance(args[5], (int, float)) else float(kwargs.get("vix_level", 15.0))
+    news          = args[6] if len(args) >= 7 and isinstance(args[6], dict) else kwargs.get("news", {})
     current_price = float(kwargs.get("current_price", 0.0))
     indicators    = _normalize_indicators(kwargs.get("indicators", {}) or {})
     ohlcv_df      = kwargs.get("ohlcv_df")

app.py

@@ -69,6 +69,7 @@ db.init_db()
 _TOP5_CACHE: dict = {}
 _TOP5_CACHE_TTL = 86400  # 24 hours — daily reset ensures new rankings based on fresh market data
 _TOP5_COMPUTING = False   # True while a background computation is in progress
+_TOP5_COMPUTING_LOCK = __import__("threading").Lock()  # guards the check-and-set on _TOP5_COMPUTING
 
 # Watchlist prediction cache — avoids re-running 12 LLM debate calls per stock on every tab switch
 _WATCHLIST_PICK_CACHE: dict = {}
@@ -137,7 +138,7 @@ def _trade_price_diagnostics(trade: dict) -> dict:
         # Include a small buffer around the trade window for context bars.
         start = (opened_date - timedelta(days=4)).strftime("%Y-%m-%d")
         end = (closed_date + timedelta(days=1)).strftime("%Y-%m-%d")
-        bars = fetch_ohlcv(ticker, start, end)
+        bars = fetch_ohlcv(ticker, period="2y")
         if bars is None or getattr(bars, "empty", True):
             return {}
 
@@ -794,9 +795,11 @@ def _start_top5_background(force_universe_refresh: bool = False) -> None:
         except Exception as exc:
             app.logger.warning("Background top5 compute failed: %s", exc)
         finally:
-            _TOP5_COMPUTING = False
+            with _TOP5_COMPUTING_LOCK:
+                _TOP5_COMPUTING = False
 
-    _TOP5_COMPUTING = True
+    with _TOP5_COMPUTING_LOCK:
+        _TOP5_COMPUTING = True
     threading.Thread(target=_run, daemon=True, name="top5-compute").start()
 
 
@@ -825,8 +828,9 @@ def top5():
     # Cache is cold — kick off background compute and return immediately so the UI
     # doesn't hang. The frontend should poll (with ?poll=1) until "computing" is gone.
     if not cache_fresh:
-        if not _TOP5_COMPUTING:
-            _start_top5_background(force_universe_refresh=force_refresh)
+        with _TOP5_COMPUTING_LOCK:
+            if not _TOP5_COMPUTING:
+                _start_top5_background(force_universe_refresh=force_refresh)
         # Serve stale result immediately while background recomputes (better UX than spinner)
         if entry and not force_refresh:
             stale_result = dict(entry["result"])
@@ -885,6 +889,7 @@ def watchlist_remove(ticker: str):
     t = _normalise(ticker)
     removed = db.remove_from_watchlist(t)
     if removed:
+        _WATCHLIST_PICK_CACHE.pop(t, None)  # evict stale prediction so re-add gets fresh data
         return jsonify({"removed": t})
     return jsonify({"error": "Not found"}), 404
 
@@ -1304,6 +1309,13 @@ def trades_open_new():
         if f not in data:
             return jsonify({"error": f"{f} is required"}), 400
 
+    try:
+        _shares_val = int(data["shares"])
+    except (TypeError, ValueError):
+        return jsonify({"error": "shares must be a positive integer"}), 400
+    if _shares_val <= 0:
+        return jsonify({"error": "shares must be > 0"}), 400
+
     ticker = _normalise(data["ticker"])
     entry_price = float(data["entry_price"])
     direction = data["direction"].upper()
@@ -1479,13 +1491,16 @@ def orders_check():
         limit = order["entry_price"]
         tol = 0.001
 
+        # Fill when live price reaches the limit — no extra tolerance at fill time.
+        # The creation-side tolerance already placed the limit below market for LONG
+        # (entry < live * 0.999), so the fill check just needs live <= entry.
         should_fill = (
-            (direction == "LONG" and live <= limit * (1 - tol)) or
-            (direction == "SHORT" and live >= limit * (1 + tol))
+            (direction == "LONG" and live <= limit) or
+            (direction == "SHORT" and live >= limit)
         )
 
         if should_fill:
-            filled_trade = db.fill_order(order["id"])
+            filled_trade = db.fill_order(order["id"], fill_price=live)
             filled_trade["fill_price"] = live
             filled.append(filled_trade)
         else:
@@ -1573,6 +1588,9 @@ def trades_close(trade_id: int):
                            close_shares=int(close_shares) if close_shares else None)
     if not trade:
         return jsonify({"error": "Trade not found"}), 404
+    if trade.get("status") not in ("CLOSED", "OPEN"):
+        # close_trade returned the trade unchanged — it was PENDING/CANCELLED, not OPEN
+        return jsonify({"error": f"Cannot close trade in status '{trade.get('status')}'"}), 409
 
     # Only generate postmortem on full close (status=CLOSED).
     if trade.get("status") == "CLOSED" and not trade.get("notes"):
@@ -1667,11 +1685,11 @@ def portfolio_review():
 
     lines = []
     for t in closed:
-        outcome = "WIN" if _safe_float(t.get("pnl_pct"), 0) >= 0 else "LOSS"
+        outcome = "WIN" if (_safe_float(t.get("pnl_pct")) or 0) >= 0 else "LOSS"
         lines.append(
             f"- {t.get('ticker','?')} | {t.get('direction','?')} | "
-            f"Entry ₹{_safe_float(t.get('entry_price'),0):.0f} → Exit ₹{_safe_float(t.get('exit_price') or t.get('current_price'),0):.0f} | "
-            f"P&L {_safe_float(t.get('pnl_pct'),0):+.2f}% | {outcome}"
+            f"Entry ₹{(_safe_float(t.get('entry_price')) or 0):.0f} → Exit ₹{(_safe_float(t.get('exit_price') or t.get('current_price')) or 0):.0f} | "
+            f"P&L {(_safe_float(t.get('pnl_pct')) or 0):+.2f}% | {outcome}"
         )
     trade_list = "\n".join(lines)
 
@@ -1705,11 +1723,7 @@ def portfolio_review():
     })
 
 
-def _safe_float(v, default=0.0):
-    try:
-        return float(v) if v is not None else default
-    except Exception:
-        return default
+# _safe_float defined above (line ~387) — this duplicate removed.
 
 
 
@@ -1809,7 +1823,9 @@ def prediction_validation():
                     )
                     entry_price = snap.get("current_price", 0)
                     if entry_price > 0 and actual_price:
-                        actual_return = round((actual_price - entry_price) / entry_price * 100, 2)
+                        raw_return = (actual_price - entry_price) / entry_price * 100
+                        # For BEARISH predictions sign-flip so positive = stock fell as predicted.
+                        actual_return = round(-raw_return if direction == "BEARISH" else raw_return, 2)
                     target_price_lo = snap.get("target_price_lo") or 0
                     target_price_hi = snap.get("target_price_hi") or 0
                     hit = _intraday_target_hit(
@@ -2441,7 +2457,14 @@ def _start_trade_monitor():
                         # Close after market close (15:31 IST) on the expiry date.
                         market_closed = now_ist.hour > 15 or (now_ist.hour == 15 and now_ist.minute >= 31)
                         if now_ist.strftime("%Y-%m-%d") >= auto_close_date and market_closed:
-                            exp_live = _current_price(trade["ticker"]) or float(trade.get("entry_price", 0))
+                            exp_live = _current_price(trade["ticker"])
+                            if not exp_live:
+                                app.logger.warning(
+                                    "Skipping auto-close for trade #%s %s — live price unavailable; "
+                                    "will retry next cycle",
+                                    trade["id"], trade["ticker"],
+                                )
+                                continue
                             closed = db.close_trade(trade["id"], exp_live)
                             if closed and not closed.get("notes"):
                                 db.save_postmortem(trade["id"], _postmortem(closed))
@@ -2462,9 +2485,9 @@ def _start_trade_monitor():
                     direction = order["direction"]
                     limit = order["entry_price"]
                     tol = 0.001
-                    if (direction == "LONG" and live <= limit * (1 - tol)) or \
-                       (direction == "SHORT" and live >= limit * (1 + tol)):
-                        db.fill_order(order["id"])
+                    if (direction == "LONG" and live <= limit) or \
+                       (direction == "SHORT" and live >= limit):
+                        db.fill_order(order["id"], fill_price=live)
                         app.logger.info(
                             "Auto-filled order #%s %s at ₹%.2f", order["id"], order["ticker"], live
                         )

data_sources.py

@@ -609,7 +609,9 @@ def fetch_live_price(ticker_ns: str, allow_delayed: bool = True) -> Optional[flo
                 if isinstance(hist.columns, pd.MultiIndex):
                     hist.columns = hist.columns.get_level_values(0)
                 closes = hist["Close"].dropna() if "Close" in hist.columns else pd.Series(dtype=float)
-                if not closes.empty:
+                # Require today's date (IST) so we don't serve yesterday's close on
+                # weekends, holidays, or after intraday bars are unavailable.
+                if not closes.empty and _is_today_ist(pd.Timestamp(closes.index[-1])):
                     return round(float(closes.iloc[-1]), 2)
         except Exception as e:
             if _is_yf_crumb_error(e):
@@ -640,7 +642,7 @@ def fetch_live_price(ticker_ns: str, allow_delayed: bool = True) -> Optional[flo
                 if isinstance(hist.columns, pd.MultiIndex):
                     hist.columns = hist.columns.get_level_values(0)
                 closes = hist["Close"].dropna() if "Close" in hist.columns else pd.Series(dtype=float)
-                if not closes.empty:
+                if not closes.empty and _is_today_ist(pd.Timestamp(closes.index[-1])):
                     return round(float(closes.iloc[-1]), 2)
         except Exception as e:
             if _is_yf_crumb_error(e):

database.py

@@ -245,14 +245,17 @@ def _migrate() -> None:
                 pass  # Columns may already exist
 
         # Remove same-day duplicate snapshots (same ticker/TF/direction/target_date created
-        # on the same UTC day). Keep the lowest id. Safe to run repeatedly — idempotent.
+        # on the same IST day). Use IST offset (+5:30) so predictions before/after UTC midnight
+        # but on the same IST trading day are correctly deduplicated.
+        # Keep the lowest id. Safe to run repeatedly — idempotent.
         try:
             conn.execute("""
                 DELETE FROM prediction_snapshots
                 WHERE id NOT IN (
                     SELECT MIN(id)
                     FROM prediction_snapshots
-                    GROUP BY ticker, timeframe, direction, validation_target_date, date(created_at)
+                    GROUP BY ticker, timeframe, direction, validation_target_date,
+                             date(created_at, '+5 hours', '+30 minutes')
                 )
             """)
         except Exception:
@@ -309,6 +312,8 @@ def _migrate() -> None:
             conn.execute("ALTER TABLE trades ADD COLUMN snapshot_id INTEGER")
         if "auto_close_date" not in cols:
             conn.execute("ALTER TABLE trades ADD COLUMN auto_close_date TEXT")
+        if "realized_pnl" not in cols:
+            conn.execute("ALTER TABLE trades ADD COLUMN realized_pnl REAL DEFAULT 0.0")
 
         if "order_type" not in cols:
             # Rebuild trades table to add order_type and extend the status CHECK.
@@ -428,14 +433,26 @@ def get_trade(trade_id: int) -> dict:
     return dict(row) if row else {}
 
 
-def fill_order(trade_id: int) -> dict:
-    """Transition a PENDING limit order to OPEN (filled)."""
+def fill_order(trade_id: int, fill_price: float | None = None) -> dict:
+    """Transition a PENDING limit order to OPEN (filled).
+
+    If fill_price is provided (the actual market price at fill time), it
+    overwrites entry_price so that P&L calculations use the real fill price
+    rather than the original limit price when the market gapped past it.
+    """
     filled_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
     with _conn() as conn:
-        conn.execute(
-            "UPDATE trades SET status = 'OPEN', opened_at = ? WHERE id = ? AND status = 'PENDING'",
-            (filled_at, trade_id),
-        )
+        if fill_price is not None:
+            conn.execute(
+                "UPDATE trades SET status = 'OPEN', opened_at = ?, entry_price = ? "
+                "WHERE id = ? AND status = 'PENDING'",
+                (filled_at, round(fill_price, 2), trade_id),
+            )
+        else:
+            conn.execute(
+                "UPDATE trades SET status = 'OPEN', opened_at = ? WHERE id = ? AND status = 'PENDING'",
+                (filled_at, trade_id),
+            )
     return get_trade(trade_id)
 
 
@@ -461,18 +478,24 @@ def merge_into_position(trade_id: int, add_shares: int, add_price: float) -> dic
     """Average a new buy into an existing OPEN position.
 
     Recalculates weighted-average entry price and increments share count.
+    The SELECT and UPDATE are in the same connection/transaction so SQLite's
+    WAL serialization prevents concurrent adds from losing shares.
     Returns the updated trade dict.
     """
-    trade = get_trade(trade_id)
-    if not trade or trade["status"] != "OPEN":
-        return trade
-    old_shares = trade["shares"]
-    old_price  = trade["entry_price"]
-    total      = old_shares + add_shares
-    avg_price  = round((old_shares * old_price + add_shares * add_price) / total, 2)
     with _conn() as conn:
+        row = conn.execute(
+            "SELECT shares, entry_price, status FROM trades WHERE id = ?", (trade_id,)
+        ).fetchone()
+        if not row or row["status"] != "OPEN":
+            return get_trade(trade_id)
+        old_shares = row["shares"]
+        old_price  = row["entry_price"]
+        total      = old_shares + add_shares
+        if total <= 0:
+            return get_trade(trade_id)
+        avg_price  = round((old_shares * old_price + add_shares * add_price) / total, 2)
         conn.execute(
-            "UPDATE trades SET shares = ?, entry_price = ? WHERE id = ?",
+            "UPDATE trades SET shares = ?, entry_price = ? WHERE id = ? AND status = 'OPEN'",
             (total, avg_price, trade_id),
         )
     return get_trade(trade_id)
@@ -506,15 +529,22 @@ def close_trade(trade_id: int, exit_price: float, close_shares: int | None = Non
     remaining = total_sh - close_sh
 
     if remaining > 0:
-        # Partial close — reduce shares, keep position OPEN, no status change.
+        # Partial close — reduce shares, accumulate realized P&L, keep OPEN.
         with _conn() as conn:
             conn.execute(
-                "UPDATE trades SET shares = ? WHERE id = ?",
-                (remaining, trade_id),
+                """
+                UPDATE trades
+                SET shares = ?,
+                    realized_pnl = COALESCE(realized_pnl, 0.0) + ?
+                WHERE id = ? AND status = 'OPEN'
+                """,
+                (remaining, round(pnl, 2), trade_id),
             )
         return get_trade(trade_id)
 
-    # Full close.
+    # Full close — guard against double-close race condition by requiring
+    # status = 'OPEN' in the UPDATE predicate. rowcount == 0 means another
+    # thread already closed this trade; skip the signal_accuracy insert.
     closed_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
     strategy  = trade.get("strategy", "")
     timeframe = trade.get("timeframe", "")
@@ -522,20 +552,21 @@ def close_trade(trade_id: int, exit_price: float, close_shares: int | None = Non
     signals   = [s.strip() for s in strategy.split(",") if s.strip()] if strategy else ["Manual"]
 
     with _conn() as conn:
-        conn.execute(
+        cur = conn.execute(
             """
             UPDATE trades
             SET exit_price = ?, closed_at = ?, status = 'CLOSED',
                 pnl = ?, pnl_pct = ?
-            WHERE id = ?
+            WHERE id = ? AND status = 'OPEN'
             """,
             (exit_price, closed_at, round(pnl, 2), round(pnl_pct, 2), trade_id),
         )
-        for sig in signals:
-            conn.execute(
-                "INSERT INTO signal_accuracy (signal, timeframe, won, pnl_pct) VALUES (?, ?, ?, ?)",
-                (sig, timeframe, won, round(pnl_pct, 2)),
-            )
+        if cur.rowcount == 1:
+            for sig in signals:
+                conn.execute(
+                    "INSERT INTO signal_accuracy (signal, timeframe, won, pnl_pct) VALUES (?, ?, ?, ?)",
+                    (sig, timeframe, won, round(pnl_pct, 2)),
+                )
 
     return get_trade(trade_id)
 

fred_data.py

@@ -156,7 +156,7 @@ def _fetch_via_fredapi() -> dict | None:
                 observation_start=end - timedelta(days=400),
                 observation_end=end,
             ).dropna()
-            if len(cpi_series) >= 12:
+            if len(cpi_series) >= 13:
                 latest_cpi = float(cpi_series.iloc[-1])
                 year_ago_cpi = float(cpi_series.iloc[-13])
                 cpi_yoy = round((latest_cpi / year_ago_cpi - 1) * 100, 2)

macro_context.py

@@ -33,7 +33,8 @@ class MacroContext:
                     close = close.iloc[:, 0]
                 frames[name] = close.rename(name)
             except Exception as e:
-                print(f"  [macro_context] Warning: could not download {ytk}: {e}")
+                import logging as _log
+                _log.getLogger(__name__).warning("macro_context: could not download %s: %s", ytk, e)
                 frames[name] = None
 
         available = {k: v for k, v in frames.items() if v is not None}
@@ -81,6 +82,11 @@ class MacroContext:
             ~feat["crude_spike"]
         )
 
+        # Drop warmup rows where indicators are NaN (rolling/pct_change warmup period).
+        # bool(NaN) == True in Python, so keeping these rows would cause the gate to
+        # silently pass as Risk-ON during the first ~20 bars of data.
+        feat = feat.dropna(subset=["sp500_trend", "usdinr_stable", "crude_spike"])
+
         # Lag all features by 1 trading day (use T-1 data to predict T direction)
         self._features = feat.shift(1)
 

predictor_core.py

@@ -17,7 +17,7 @@ Public API:
 """
 
 from __future__ import annotations
-import sys, os, warnings, math, time, logging
+import sys, os, warnings, math, time, logging, threading
 warnings.filterwarnings("ignore")
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -303,7 +303,7 @@ def _load_live_strategy_stats() -> dict:
             # avg_pnl = avg_win * (wr - lr / rr)  → solve for avg_win
             denom = win_rate - (lr / old_rr)
             if abs(denom) > 1e-6:
-                new_win  = max(0.5, avg_pnl / denom)
+                new_win  = max(0.5, min(avg_pnl / denom, 20.0))  # cap at 20× so corrupt rows can't inflate EV
                 new_loss = max(0.1, new_win / old_rr)
                 stats[sig] = (round(win_rate, 3), round(new_win, 2), round(new_loss, 2))
     except Exception:
@@ -358,14 +358,16 @@ def clear_runtime_caches() -> dict:
 # ── LOG THROTTLE (avoid per-ticker warning spam) ────────────────────────────
 _LAST_NIFTY_WARN_TS = 0.0
 _NIFTY_WARN_TTL = 300  # seconds
+_NIFTY_WARN_LOCK = threading.Lock()
 
 
 def _warn_nifty_unavailable_once() -> None:
     global _LAST_NIFTY_WARN_TS
     now = time.time()
-    if now - _LAST_NIFTY_WARN_TS >= _NIFTY_WARN_TTL:
-        logging.warning("Nifty market data unavailable — Nifty-dependent signals skipped (throttled)")
-        _LAST_NIFTY_WARN_TS = now
+    with _NIFTY_WARN_LOCK:
+        if now - _LAST_NIFTY_WARN_TS >= _NIFTY_WARN_TTL:
+            logging.warning("Nifty market data unavailable — Nifty-dependent signals skipped (throttled)")
+            _LAST_NIFTY_WARN_TS = now
 
 
 def _get_market_cache(period: str = "1y"):
@@ -447,7 +449,7 @@ def _get_nifty_gate() -> tuple[bool, str]:
     making the current EMA200 level accurate. Requires 3 of the last 5 closes
     below the warm EMA200 to trigger CAUTION — majority-of-week rule filters noise."""
     nifty_c, _ = _get_market_cache("2y")  # 2y → EMA200 is fully warmed up
-    if nifty_c is None or len(nifty_c) == 0:
+    if nifty_c is None or len(nifty_c) < 3:
         return True, "Nifty data unavailable — assuming OK"
     ema200_s = nifty_c.ewm(span=200).mean()
     ema200   = float(ema200_s.iloc[-1])
@@ -471,9 +473,19 @@ def _get_macro_gate() -> tuple[bool, str]:
     start = (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%d")
     mc    = MacroContext()
     mc.load(start, end)
-    today = pd.Timestamp.now().normalize()
-    feat  = mc.get(today)
-    ok    = feat.get("global_risk_on", True)
+    # _features is shifted by 1 day so today's date is never in the index.
+    # Use build_mask with ffill to get the most recent available T-1 row.
+    _today_ts = pd.Timestamp.now().normalize()
+    _mask = mc.build_mask(pd.DatetimeIndex([_today_ts]))
+    ok    = bool(_mask.iloc[0]) if not _mask.empty else True
+    # Re-fetch the full feature row for the description.
+    feat  = mc.get(_today_ts - pd.Timedelta(days=1))
+    if not feat:
+        # Fallback: scan backwards up to 5 days for a valid row.
+        for _d in range(1, 6):
+            feat = mc.get(_today_ts - pd.Timedelta(days=_d))
+            if feat:
+                break
     parts = []
     if not feat.get("sp500_trend", True):
         parts.append("S&P500 weak")
@@ -495,8 +507,10 @@ def get_earnings_status(ticker: str, window: int = 5) -> dict:
             return {"next_date": None, "days_away": None, "in_blackout": False, "warning": None}
 
         today = datetime.now().date()
-        # Look at upcoming earnings (within next 30 days) AND recent past (within 5 days)
-        for edate in ed.index:
+        # Sort descending so the nearest upcoming date is encountered first.
+        # yfinance returns earnings_dates in undocumented order — sorting avoids
+        # returning a distant date when a near one is also in the window.
+        for edate in sorted(ed.index, key=lambda d: abs((d.date() - today).days) if hasattr(d, "date") else abs((pd.Timestamp(d).date() - today).days)):
             edate_d = edate.date() if hasattr(edate, "date") else pd.Timestamp(edate).date()
             days_away = (edate_d - today).days
             if -window <= days_away <= 30:
@@ -1466,8 +1480,8 @@ def predict_stock_v2(ticker: str, start_date: str, end_date: str,
     elif _ml_score < 50:
         _entry_buffer = _entry_buffer * 1.2  # 20% more conservative
 
-    _is_actionable_buy  = direction in ("BUY", "STRONG BUY") and no_trade_reason is None
-    _is_actionable_sell = direction in ("SELL", "BEARISH")    and no_trade_reason is None
+    _is_actionable_buy  = direction in ("BULLISH", "SLIGHTLY BULLISH") and no_trade_reason is None
+    _is_actionable_sell = direction in ("SELL", "BEARISH")             and no_trade_reason is None
     if _is_actionable_buy:
         entry_price = round(price * (1 + _entry_buffer), 2)
         entry_basis = "est_open_conservative" if _ml_score < 50 else "est_open"
@@ -1518,7 +1532,7 @@ def predict_stock_v2(ticker: str, start_date: str, end_date: str,
     # ── Phase 6: Price targets (Camarilla / ATR / PDH) ──────────────────────────
     try:
         from price_targets import get_price_targets
-        _strategy_bias = "BULLISH" if direction in ("BUY", "STRONG BUY", "BULLISH") else (
+        _strategy_bias = "BULLISH" if direction in ("BULLISH", "SLIGHTLY BULLISH") else (
                   "BEARISH" if direction in ("SELL", "BEARISH") else "NEUTRAL")
         price_targets_dict = get_price_targets(ticker, sc, sh, sl, _strategy_bias, confidence)
     except Exception as _pt_err:

trial_run.py

@@ -63,7 +63,7 @@ def adx_s(h, l, c, n=14):
     up = h.diff(); dn = -l.diff()
     pdm = up.where((up > dn) & (up > 0), 0.0)
     ndm = dn.where((dn > up) & (dn > 0), 0.0)
-    at  = atr(h, l, c, n)
+    at  = atr(h, l, c, n).replace(0, np.nan)  # guard: frozen stocks have ATR=0 → inf/NaN in ADX
     pdi = 100 * pdm.ewm(com=n-1).mean() / at
     ndi = 100 * ndm.ewm(com=n-1).mean() / at
     dx  = 100 * (pdi - ndi).abs() / (pdi + ndi).replace(0, np.nan)