chore: post-audit tidies (CB flag path, correlation logging, no-touch list, dynamic-stop tiers) (#181)

Five small fixes from the comprehensive repo audit that don't justify
their own PRs:

H1: resilience._send_circuit_breaker_alert wrote the fallback flag file
    via a re-derived Path(__file__) instead of the module-level
    _CB_FLAG_PATH constant. Test fixtures that monkeypatch the constant
    (tests/conftest.py::_clear_circuit_breaker_flag) didn't redirect
    this fallback, so a failing alert in tests would touch the
    production data/ directory. Now uses the constant — fixture takes
    effect.

L5: correlation._fetch_close_prices silently dropped symbols missing
    from the batch response (delisted, no data in window). Now logs
    a single WARNING listing the dropped symbols and the count, so
    operators can see when the guard is degraded.

Add strategies/state_db.py to AGENT_NO_TOUCH_FILES. state_db owns the
SQLite schema and atomic-transaction patterns for ALL trading state;
an agent-authored migration could corrupt the live DB across the
no-rollback boundary. New test_contains_state_db assertion locks it.

M6: pipeline_backtest._dynamic_trailing_stop_pct hardcoded the
    100/75/50/37.5 multipliers instead of reading TRAILING_STOP_TIERS
    from constants. The live exit manager (position_manager.py) reads
    the constant, so changing tier values in constants would silently
    desync the backtest from production. Now both consume the same
    source of truth.

H4: scan_with_sentiment.py doesn't apply the pair-wise correlation
    guard. Documented as intentional — the watchlist is small (~4
    curated tickers) and operators control composition directly. The
    orchestrator (screener/momentum entry paths) keeps the guard
    because its candidate universe is wide and dynamic.

Test plan:
  - 2825/2825 pass (added 1 test_contains_state_db)
  - Per-file coverage floors OK (69 files; default 85%)
  - Total coverage 95.48%

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: costajohnt

scripts/pipeline_backtest.py

@@ -48,7 +48,12 @@
     compute_max_drawdown,
     compute_sharpe_ratio,
 )
-from strategies.constants import DEFAULT_TAKE_PROFIT_PCT, DEFAULT_TRAILING_STOP_PCT
+from strategies.constants import (
+    DEFAULT_TAKE_PROFIT_PCT,
+    DEFAULT_TRAILING_STOP_PCT,
+    TRAILING_STOP_TIERS,
+    TRAILING_STOP_UNDERWATER_MULTIPLIER,
+)
 from strategies.enums import Action
 from strategies.log import get_logger
 from strategies.signal_scorer import (
@@ -214,23 +219,19 @@ def is_on_cooldown(self, symbol: str, date: datetime) -> bool:
     def _dynamic_trailing_stop_pct(self, pnl_pct: float) -> float:
         """Return effective trailing stop % based on unrealized P&L.
 
-        Mirrors position_manager.dynamic_trailing_stop_pct:
-          P&L < 0%:   100%  of base (full room)
-          P&L 0-3%:    75%  of base
-          P&L 3-5%:    50%  of base
-          P&L >= 5%:   37.5% of base
+        Mirrors position_manager.dynamic_trailing_stop_pct — both read the
+        same tiers from strategies.constants so the backtest's behavior
+        cannot drift from the live exit manager (a previous bug had
+        hard-coded 100/75/50/37.5 multipliers here that silently fell out
+        of sync when constants changed).
         """
         if not self.use_dynamic_stop:
             return self.trailing_stop_pct
 
-        if pnl_pct < 0:
-            return self.trailing_stop_pct
-        elif pnl_pct < 3:
-            return self.trailing_stop_pct * 0.75
-        elif pnl_pct < 5:
-            return self.trailing_stop_pct * 0.50
-        else:
-            return self.trailing_stop_pct * 0.375
+        for threshold, multiplier in TRAILING_STOP_TIERS:
+            if pnl_pct >= threshold:
+                return self.trailing_stop_pct * multiplier
+        return self.trailing_stop_pct * TRAILING_STOP_UNDERWATER_MULTIPLIER
 
     # --- Entry ---
 

scripts/scan_with_sentiment.py

@@ -183,6 +183,13 @@ def main():
     results = []
     tech_errors = 0
 
+    # Note: this watchlist scan does NOT apply the pair-wise correlation
+    # guard (see strategies/correlation.py). The watchlist is small (~4
+    # curated tickers) and operators control its composition directly,
+    # so over-correlation is managed at the watchlist-design level rather
+    # than enforced at runtime. The orchestrator's screener/momentum paths
+    # use the correlation guard because their candidate universe is wide
+    # and dynamic.
     for entry in WATCHLIST:
         # 1. Technical analysis
         strategy = MeanReversionStrategy(

strategies/correlation.py

@@ -80,21 +80,34 @@ def _fetch_close_prices(
         return {}
 
     closes: dict[str, pd.Series] = {}
+    dropped: list[str] = []
     if "symbol" in df.index.names:
         for sym in symbols:
             try:
                 sym_df = df.xs(sym, level="symbol")
                 if sym_df.empty:
+                    dropped.append(sym)
                     continue
                 closes[sym] = sym_df["close"].astype(float)
             except KeyError:
                 # Symbol absent from the batch response (delisted, no
-                # data in window, etc.). Drop silently.
+                # data in window, etc.). Track and log so the missing
+                # symbol isn't invisible to the caller.
+                dropped.append(sym)
                 continue
     else:
         # Single-symbol response shape (rare; normalise just in case).
         if len(symbols) == 1 and not df.empty:
             closes[symbols[0]] = df["close"].astype(float)
+        else:
+            dropped.extend(symbols)
+
+    if dropped:
+        logger.warning(
+            "Correlation guard: %d/%d symbols had no bars in the window "
+            "(%s) — excluded from correlation matrix.",
+            len(dropped), len(symbols), ", ".join(sorted(dropped)),
+        )
 
     return closes
 

strategies/resilience.py

@@ -156,11 +156,14 @@ def _send_circuit_breaker_alert():
         )
     except Exception as alert_err:
         logger.critical("CIRCUIT BREAKER: Could not send alert (%s) — check notify.py", alert_err)
-        # Write flag file so pipeline can detect the trip even if alerts fail
+        # Write flag file so pipeline can detect the trip even if alerts fail.
+        # Use the module-level constant so test fixtures that monkeypatch
+        # _CB_FLAG_PATH (e.g. tests/conftest.py::_clear_circuit_breaker_flag)
+        # also redirect this fallback write — otherwise a failing alert in a
+        # test would touch the production flag path.
         try:
-            cb_path = Path(__file__).resolve().parent.parent / "data" / "circuit_breaker.flag"
-            cb_path.parent.mkdir(parents=True, exist_ok=True)
-            cb_path.write_text(str(time.time()))
+            _CB_FLAG_PATH.parent.mkdir(parents=True, exist_ok=True)
+            _CB_FLAG_PATH.write_text(str(time.time()))
         except OSError as file_err:
             logger.critical("CIRCUIT BREAKER: Cannot write flag file (%s) — trip has NO persistent record", file_err)
 

strategies/safety.py

@@ -48,6 +48,11 @@
     "strategies/risk_guard.py",
     "strategies/resilience.py",
     "strategies/safety.py",
+    # state_db owns SQLite schema + atomic transaction patterns for ALL
+    # trading state (HWMs, cooldowns, partial profits, equity snapshots,
+    # PDT, EDGAR cache). A schema migration written by the agent could
+    # corrupt the live DB across the no-rollback boundary.
+    "strategies/state_db.py",
     "scripts/strategy_agent.py",
     "scripts/agent_tools.py",
     "scripts/agent_reads.py",

tests/test_safety.py

@@ -82,6 +82,11 @@ def test_contains_risk_guard(self):
     def test_contains_resilience(self):
         assert "strategies/resilience.py" in AGENT_NO_TOUCH_FILES
 
+    def test_contains_state_db(self):
+        """state_db.py owns SQLite schema + atomic transactions for ALL
+        trading state — must be agent-protected."""
+        assert "strategies/state_db.py" in AGENT_NO_TOUCH_FILES
+
     def test_contains_agent_tools(self):
         assert "scripts/agent_tools.py" in AGENT_NO_TOUCH_FILES