news: beginner-language summary, 4 h headline window, drop neutral headlines; --news log report
Verdict from the 3-week signal log (backtest/research/news_sentiment_value_2026-09-04.md): news_sentiment is a trailing-24h echo of price (r +0.67 with the past day, +0.03 with the next; most bullish 2 h after the 08-25 top, bearish 44 h later; most bearish at the 09-02 low, bullish 16 h later). Per-headline LLM labels are correct; the 12 h pool and the averaging are what destroy the information. Both gating flags stay False. - news_engine: _LLM_SUMMARY_SYSTEM rewritten for someone who has never traded (no acronyms, two worked examples); summary payload newest-first with the platform's own gold direction; reply cap 300 -> 400 chars. - config: headline_max_age_seconds 43200 -> 14400; RSS items without a parsable pubDate are dropped (they used to be stamped "now" and never aged out). - news_engine.get_fundamental_context drops headlines whose every relevant score is inside +-0.1 (_is_neutral_headline) before the per-currency mean; news_sentiment from this date is a mean over directional <=4 h headlines (log comparisons split here). - static: dashboard copy for the 4 h pool; news_summary.php prompt gets the same style rule. - analyze_signal_log: --news (under --stored-candles) prints sentiment lead/lag with split halves, sentiment at the window high/low with flip lags, and outcome buckets by direction x news x macro stance; LogRow gains news_sentiment/event_risk; _row_index shared. - tests: 406 -> 414 (age window, undated drop, neutral filter, summary prompt/payload/cap, --news helpers, load_rows round-trip); RSS fixtures now carry a pubDate. - docs: user/beginner guides, CLAUDE.md, deploy README. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0169BgFtWjp1SEAa5392Q9jH
This commit is contained in:
Harwinder Sekhon
+279
-9
@@ -38,6 +38,14 @@ Three lenses:
|
||||
the "opportunity" (first event per tf+direction cluster) and
|
||||
one-position-per-timeframe views.
|
||||
|
||||
6. NEWS SENTIMENT (--news, with --stored-candles) — is the logged
|
||||
news_sentiment score a leading or a trailing signal? Hourly mean
|
||||
sentiment vs past/forward price moves at 1/4/8/24 h (split-half on
|
||||
the forward number), the daily open->close read, what it said at the
|
||||
window's high and low, and resolved-event R bucketed by
|
||||
direction x news bucket x macro-persona stance. Warning-only feature:
|
||||
nothing here gates a signal.
|
||||
|
||||
The DB is opened read-only (mode=ro — WAL-visible; the older immutable=1
|
||||
URI silently missed the newest rows). Bar data for the default replay
|
||||
comes from the active provider via the backtester's fetch_history
|
||||
@@ -50,6 +58,7 @@ Usage:
|
||||
python analyze_signal_log.py --no-fetch # cache only
|
||||
python analyze_signal_log.py --parity # surface parity only
|
||||
python analyze_signal_log.py --stored-candles # DB-only replay
|
||||
python analyze_signal_log.py --stored-candles --news # + news lens
|
||||
"""
|
||||
|
||||
import argparse
|
||||
@@ -104,6 +113,8 @@ class LogRow:
|
||||
advisory_status: str = ""
|
||||
advisory_r: Optional[float] = None
|
||||
scan_kind: str = ""
|
||||
news_sentiment: Optional[float] = None
|
||||
event_risk: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -132,7 +143,14 @@ _COLS = ("id, logged_at, source, timeframe, signal, score, quality,"
|
||||
# Columns added by the 2026-09 observability work; absent on older DBs.
|
||||
_OPTIONAL_COLS = ("gate_reason", "actionable", "config_fingerprint",
|
||||
"advisory_event_id", "advisory_outcome_status",
|
||||
"advisory_r_multiple", "scan_kind")
|
||||
"advisory_r_multiple", "scan_kind", "news_sentiment",
|
||||
"event_risk")
|
||||
|
||||
# The aggregator's own "News:" reason threshold — the band outside which a
|
||||
# logged sentiment score is called bullish/bearish for gold.
|
||||
NEWS_BULL = 0.1
|
||||
NEWS_BEAR = -0.1
|
||||
NEWS_LAGS_H = (1, 4, 8, 24)
|
||||
|
||||
|
||||
def open_readonly(db_path: str) -> sqlite3.Connection:
|
||||
@@ -184,6 +202,10 @@ def load_rows(db_path: str) -> List[LogRow]:
|
||||
advisory_status=extra.get("advisory_outcome_status") or "",
|
||||
advisory_r=extra.get("advisory_r_multiple"),
|
||||
scan_kind=extra.get("scan_kind") or "",
|
||||
news_sentiment=(float(extra["news_sentiment"])
|
||||
if extra.get("news_sentiment") is not None
|
||||
else None),
|
||||
event_risk=extra.get("event_risk") or "",
|
||||
))
|
||||
return out
|
||||
finally:
|
||||
@@ -232,6 +254,12 @@ def cluster_setup_rows(
|
||||
return events
|
||||
|
||||
|
||||
def _row_index(events: Sequence[SignalEvent]) -> Dict[int, SignalEvent]:
|
||||
"""row id -> the SignalEvent it belongs to (every row of an event
|
||||
shares that event's replayed outcome)."""
|
||||
return {r.id: ev for ev in events for r in ev.rows}
|
||||
|
||||
|
||||
# ── Trade construction + replay (backtester realism, reused code) ────
|
||||
|
||||
|
||||
@@ -380,10 +408,7 @@ def panel_scorecard(
|
||||
Verdicted rows with a setup join their SignalEvent outcome; rows
|
||||
without a setup are scored on the +4h forward return instead.
|
||||
"""
|
||||
by_row: Dict[int, SignalEvent] = {}
|
||||
for ev in events:
|
||||
for r in ev.rows:
|
||||
by_row[r.id] = ev
|
||||
by_row = _row_index(events)
|
||||
|
||||
out = []
|
||||
for row in rows:
|
||||
@@ -725,10 +750,7 @@ def persona_scorecard(rows: Sequence[LogRow], events: Sequence[SignalEvent],
|
||||
db_path: str) -> List[dict]:
|
||||
"""Per persona × stance: n, mean R and win rate of the underlying
|
||||
event (needs panel_takes_json — read straight from the DB)."""
|
||||
by_row: Dict[int, SignalEvent] = {}
|
||||
for ev in events:
|
||||
for r in ev.rows:
|
||||
by_row[r.id] = ev
|
||||
by_row = _row_index(events)
|
||||
con = open_readonly(db_path)
|
||||
agg: Dict[Tuple[str, str], List[float]] = {}
|
||||
try:
|
||||
@@ -753,6 +775,243 @@ def persona_scorecard(rows: Sequence[LogRow], events: Sequence[SignalEvent],
|
||||
for (p, s), v in sorted(agg.items())]
|
||||
|
||||
|
||||
# ── News sentiment lead/lag ───────────────────────────────────────────
|
||||
|
||||
|
||||
def _hourly_sentiment(rows: Sequence[LogRow]) -> pd.Series:
|
||||
"""Mean logged news_sentiment per UTC hour (priced rows only)."""
|
||||
buckets: Dict[pd.Timestamp, List[float]] = {}
|
||||
for r in rows:
|
||||
if not r.price or r.news_sentiment is None:
|
||||
continue
|
||||
ts = pd.Timestamp(r.logged_at)
|
||||
ts = ts.tz_localize("UTC") if ts.tzinfo is None else ts.tz_convert("UTC")
|
||||
buckets.setdefault(ts.floor("h"), []).append(float(r.news_sentiment))
|
||||
if not buckets:
|
||||
return pd.Series(dtype=float)
|
||||
return pd.Series({k: sum(v) / len(v)
|
||||
for k, v in buckets.items()}).sort_index()
|
||||
|
||||
|
||||
def _stream_frame(streams: Dict[str, List[dict]], tf: str) -> pd.DataFrame:
|
||||
"""Stored bars for one timeframe as an OHLC frame indexed by UTC ts."""
|
||||
bars = streams.get(tf) or []
|
||||
if not bars:
|
||||
return pd.DataFrame()
|
||||
df = pd.DataFrame(bars)
|
||||
if "timestamp" not in df.columns or "close" not in df.columns:
|
||||
return pd.DataFrame()
|
||||
df.index = pd.to_datetime(df["timestamp"], utc=True)
|
||||
keep = [c for c in ("open", "high", "low", "close") if c in df.columns]
|
||||
return df[keep].astype(float).sort_index()
|
||||
|
||||
|
||||
def _corr(a: pd.Series, b: pd.Series) -> Optional[float]:
|
||||
"""Pearson correlation of the overlapping, finite pairs (None if NaN)."""
|
||||
try:
|
||||
c = float(a.corr(b))
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return None if math.isnan(c) else c
|
||||
|
||||
|
||||
def news_lead_lag(rows: Sequence[LogRow], streams: Dict[str, List[dict]],
|
||||
tf: str = "1h") -> dict:
|
||||
"""Does the logged sentiment LEAD price, or only follow it?
|
||||
|
||||
For each lag k (hours) the hourly mean sentiment is correlated with
|
||||
the PAST move (open_t - open_{t-k}) and the FORWARD move
|
||||
(close_{t+k-1} - open_t) — the scan in hour H sees the bar that
|
||||
closed at H, so the forward window starts at bar H's open. A high
|
||||
past-correlation with a ~zero forward-correlation means the score is
|
||||
a trailing return in disguise. Split-half on the forward number
|
||||
guards against reading one regime as an edge.
|
||||
"""
|
||||
sent = _hourly_sentiment(rows)
|
||||
df = _stream_frame(streams, tf)
|
||||
if sent.empty or df.empty:
|
||||
return {"n_hours": 0}
|
||||
joined = df.join(pd.DataFrame({"sent": sent}), how="inner").dropna(
|
||||
subset=["sent"])
|
||||
n = len(joined)
|
||||
if n < 8:
|
||||
return {"n_hours": n}
|
||||
s = joined["sent"]
|
||||
half = n // 2
|
||||
first, second = joined.index[:half], joined.index[half:]
|
||||
corr_past, corr_fwd, corr_split = {}, {}, {}
|
||||
for k in NEWS_LAGS_H:
|
||||
past = (df["open"] - df["open"].shift(k)).reindex(joined.index)
|
||||
fwd = (df["close"].shift(-(k - 1)) - df["open"]).reindex(joined.index)
|
||||
corr_past[k] = _corr(s, past)
|
||||
corr_fwd[k] = _corr(s, fwd)
|
||||
corr_split[k] = (_corr(s.loc[first], fwd.loc[first]),
|
||||
_corr(s.loc[second], fwd.loc[second]))
|
||||
day_sent = sent.groupby(sent.index.floor("D")).mean()
|
||||
day = df.resample("1D").agg({"open": "first", "close": "last"}).dropna()
|
||||
move = day["close"] - day["open"]
|
||||
# .corr aligns on the index; shifting over the BAR index keeps
|
||||
# "previous/next" meaning the previous/next day that actually traded.
|
||||
daily = {"same": _corr(day_sent, move),
|
||||
"prev": _corr(day_sent, move.shift(1)),
|
||||
"next": _corr(day_sent, move.shift(-1))}
|
||||
return {"n_hours": n, "corr_past": corr_past, "corr_fwd": corr_fwd,
|
||||
"corr_fwd_split": corr_split, "daily": daily}
|
||||
|
||||
|
||||
def news_extremes(rows: Sequence[LogRow], streams: Dict[str, List[dict]],
|
||||
tf: str = "1h") -> dict:
|
||||
"""What the sentiment said at the window's high and low.
|
||||
|
||||
Plus how long AFTER the top the first bearish hour arrived (and vice
|
||||
versa at the bottom) — a turn called late is a trailing indicator.
|
||||
"""
|
||||
df = _stream_frame(streams, tf)
|
||||
if df.empty or "high" not in df.columns or "low" not in df.columns:
|
||||
return {}
|
||||
sent = _hourly_sentiment(rows)
|
||||
hi_ts = df["high"].idxmax()
|
||||
lo_ts = df["low"].idxmin()
|
||||
hi_px = float(df["high"].loc[hi_ts])
|
||||
lo_px = float(df["low"].loc[lo_ts])
|
||||
|
||||
def _at(ts):
|
||||
v = sent.get(ts)
|
||||
return None if v is None or (isinstance(v, float) and math.isnan(v)) \
|
||||
else float(v)
|
||||
|
||||
def _first(from_ts, ref_px, sign):
|
||||
for ts, val in sent.items():
|
||||
if ts < from_ts:
|
||||
continue
|
||||
if (val <= NEWS_BEAR) if sign < 0 else (val >= NEWS_BULL):
|
||||
close = df["close"].asof(ts)
|
||||
if close is None or (isinstance(close, float)
|
||||
and math.isnan(close)):
|
||||
continue
|
||||
return {"lag_h": (ts - from_ts).total_seconds() / 3600.0,
|
||||
"price_move": float(close) - ref_px}
|
||||
return None
|
||||
|
||||
return {
|
||||
"high": {"ts": hi_ts.isoformat(), "price": hi_px,
|
||||
"sentiment": _at(hi_ts)},
|
||||
"low": {"ts": lo_ts.isoformat(), "price": lo_px,
|
||||
"sentiment": _at(lo_ts)},
|
||||
"bear_after_high": _first(hi_ts, hi_px, -1),
|
||||
"bull_after_low": _first(lo_ts, lo_px, +1),
|
||||
}
|
||||
|
||||
|
||||
def news_outcome_buckets(rows: Sequence[LogRow],
|
||||
events: Sequence[SignalEvent],
|
||||
db_path: str) -> List[dict]:
|
||||
"""Resolved events bucketed by (direction, news bucket, macro stance).
|
||||
|
||||
The macro persona is the only voice the 2026-09 scorecard found
|
||||
informative, so it rides along as a third axis here.
|
||||
"""
|
||||
by_row = _row_index(events)
|
||||
macro: Dict[int, str] = {}
|
||||
con = open_readonly(db_path)
|
||||
try:
|
||||
for rid, takes in con.execute(
|
||||
"SELECT id, panel_takes_json FROM signals"
|
||||
" WHERE panel_takes_json IS NOT NULL ORDER BY id"):
|
||||
ev = by_row.get(rid)
|
||||
if ev is None or ev.representative.id in macro:
|
||||
continue
|
||||
try:
|
||||
parsed = json.loads(takes)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
stance = "none"
|
||||
for t in parsed or []:
|
||||
if isinstance(t, dict) and t.get("persona") == "macro":
|
||||
stance = t.get("stance") or "none"
|
||||
break
|
||||
macro[ev.representative.id] = stance
|
||||
finally:
|
||||
con.close()
|
||||
|
||||
agg: Dict[Tuple[str, str, str], List[float]] = {}
|
||||
for ev in events:
|
||||
if ev.r_headline is None:
|
||||
continue
|
||||
rep = ev.representative
|
||||
s = rep.news_sentiment
|
||||
if s is None:
|
||||
bucket = "neutral"
|
||||
elif s >= NEWS_BULL:
|
||||
bucket = "bull"
|
||||
elif s <= NEWS_BEAR:
|
||||
bucket = "bear"
|
||||
else:
|
||||
bucket = "neutral"
|
||||
direction = (rep.setup or {}).get("direction") or "?"
|
||||
agg.setdefault((direction, bucket, macro.get(rep.id, "none")),
|
||||
[]).append(ev.r_headline)
|
||||
return [{"direction": d, "bucket": b, "macro": m, "n": len(v),
|
||||
"sum_r": sum(v), "mean_r": sum(v) / len(v),
|
||||
"win_rate": sum(1 for x in v if x > 0) / len(v)}
|
||||
for (d, b, m), v in sorted(agg.items())]
|
||||
|
||||
|
||||
def _fmt_c(c: Optional[float], nd: int = 3) -> str:
|
||||
return " n/a " if c is None else f"{c:+.{nd}f}"
|
||||
|
||||
|
||||
def print_news(lead_lag: dict, extremes: dict, buckets: Sequence[dict],
|
||||
tf: str = "1h") -> None:
|
||||
print("\n── News sentiment (warning-only; hourly mean vs stored "
|
||||
+ tf + " bars) " + "─" * 10)
|
||||
n = lead_lag.get("n_hours", 0)
|
||||
if "corr_past" not in lead_lag:
|
||||
print(f" only {n} joined hour(s) of logged sentiment — "
|
||||
"need 8+ for a correlation read")
|
||||
else:
|
||||
print(f" {n} hours with a logged sentiment joined to a bar")
|
||||
for k in NEWS_LAGS_H:
|
||||
a, b = lead_lag["corr_fwd_split"].get(k, (None, None))
|
||||
print(f" past {k:2d}h {_fmt_c(lead_lag['corr_past'].get(k))}"
|
||||
f" fwd {k:2d}h {_fmt_c(lead_lag['corr_fwd'].get(k))}"
|
||||
f" (halves {_fmt_c(a, 2)} / {_fmt_c(b, 2)})")
|
||||
d = lead_lag.get("daily") or {}
|
||||
print(f" daily open->close: same day {_fmt_c(d.get('same'))}"
|
||||
f" previous {_fmt_c(d.get('prev'))}"
|
||||
f" next {_fmt_c(d.get('next'))}")
|
||||
|
||||
for side, key, tag in (("high", "bear_after_high", "bearish"),
|
||||
("low", "bull_after_low", "bullish")):
|
||||
info = (extremes or {}).get(side)
|
||||
if not info:
|
||||
continue
|
||||
sv = info.get("sentiment")
|
||||
head = (f" {side:4} {info['ts'][:16]}Z ${info['price']:.1f} "
|
||||
f"sentiment {_fmt_c(sv, 2)}")
|
||||
nxt = (extremes or {}).get(key)
|
||||
if nxt:
|
||||
print(f"{head} → first {tag} hour +{nxt['lag_h']:.0f} h later, "
|
||||
f"price {nxt['price_move']:+.1f}")
|
||||
else:
|
||||
print(f"{head} → no {tag} hour logged after it")
|
||||
|
||||
if buckets:
|
||||
print(f" outcome buckets ({sum(b['n'] for b in buckets)} resolved "
|
||||
"events):")
|
||||
print(f" {'dir':5} {'news':7} {'macro':8} {'n':>3} {'total':>8} "
|
||||
f"{'mean':>8} {'win':>5}")
|
||||
for b in buckets:
|
||||
print(f" {b['direction']:5} {b['bucket']:7} {b['macro']:8} "
|
||||
f"{b['n']:3d} {_fmt_r(b['sum_r']):>8} "
|
||||
f"{_fmt_r(b['mean_r']):>8} "
|
||||
f"{100 * b['win_rate']:4.0f}%")
|
||||
else:
|
||||
print(" no resolved events to bucket yet")
|
||||
print(" Caveat: sentiment ≈ trailing 24h return; buckets inherit the "
|
||||
"trend confound.")
|
||||
|
||||
|
||||
def print_report(
|
||||
rows: List[LogRow],
|
||||
events: List[SignalEvent],
|
||||
@@ -892,8 +1151,15 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
|
||||
ap.add_argument("--stored-candles", action="store_true",
|
||||
help="replay on the candle streams stored in the DB "
|
||||
"(no network) instead of provider history")
|
||||
ap.add_argument("--news", action="store_true",
|
||||
help="news-sentiment lead/lag + outcome buckets "
|
||||
"(requires --stored-candles)")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.news and not args.stored_candles:
|
||||
print("--news requires --stored-candles")
|
||||
return 2
|
||||
|
||||
rows = load_rows(args.db)
|
||||
if not rows:
|
||||
print(f"No rows in {args.db}")
|
||||
@@ -923,6 +1189,10 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
|
||||
for p in persona_scorecard(rows, events, args.db):
|
||||
print(f" {p['persona']:11} {p['stance']:8} n={p['n']:3d} "
|
||||
f"mean {p['mean_r']:+.3f}R win {100 * p['win_rate']:.0f}%")
|
||||
if args.news:
|
||||
print_news(news_lead_lag(rows, streams),
|
||||
news_extremes(rows, streams),
|
||||
news_outcome_buckets(rows, events, args.db))
|
||||
print_parity(parity_report(rows))
|
||||
if args.csv:
|
||||
export_csv(events, args.csv)
|
||||
|
||||
Reference in New Issue
Block a user