harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

signal_log: persist market snapshots

Browse Source
This commit is contained in:
Harwinder Sekhon
2026-08-18 15:24:37 -04:00
parent 71eb703355
commit 410bf5ebc4
5 changed files with 219 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
signal_log.py
+139 -16
View File
@@ -68,6 +68,8 @@ CREATE TABLE IF NOT EXISTS signals (
news_summary TEXT,
events_json TEXT,
headlines_json TEXT,
indicators_json TEXT,
candles_json TEXT,
panel_verdict TEXT,
panel_confidence REAL,
panel_rationale TEXT,
@@ -83,17 +85,28 @@ CREATE TABLE IF NOT EXISTS outlooks (
setup_forming TEXT,
confidence REAL,
watch TEXT,
note TEXT
note TEXT,
news_json TEXT,
indicators_json TEXT,
candles_json TEXT
);
"""
# Additive migration: older DBs created before the panel columns existed
# get them added here. Idempotent — skips columns already present.
_PANEL_COLS = (
# Additive migration for columns introduced after the original schema.
# Idempotent — skips columns already present.
_ADDITIVE_SIGNAL_COLS = (
("panel_verdict", "TEXT"),
("panel_confidence", "REAL"),
("panel_rationale", "TEXT"),
("panel_takes_json", "TEXT"),
("indicators_json", "TEXT"),
("candles_json", "TEXT"),
)
_ADDITIVE_OUTLOOK_COLS = (
("news_json", "TEXT"),
("indicators_json", "TEXT"),
("candles_json", "TEXT"),
)
_INSERT = """
@@ -103,14 +116,16 @@ INSERT INTO signals (
mtf_json, setup_json, reasons_json, warnings_json,
event_risk, event_description, news_sentiment, news_description,
news_summary, events_json, headlines_json,
indicators_json, candles_json,
panel_verdict, panel_confidence, panel_rationale, panel_takes_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
_OUTLOOK_INSERT = """
INSERT INTO outlooks (
logged_at, surface, bias, setup_forming, confidence, watch, note
) VALUES (?, ?, ?, ?, ?, ?, ?)
logged_at, surface, bias, setup_forming, confidence, watch, note,
news_json, indicators_json, candles_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
@@ -165,8 +180,16 @@ def _headline_to_dict(h: Any) -> Dict[str, Any]:
"source": getattr(h, "source", None),
"timestamp": _iso(getattr(h, "timestamp", None)),
"score": _finite(getattr(h, "sentiment_score", None)),
"score_by_currency": {
str(currency): _finite(score)
for currency, score in (
getattr(h, "sentiment_by_currency", None) or {}).items()
},
"currencies": list(getattr(h, "currencies_mentioned", None) or []),
"keywords": list(getattr(h, "keywords", None) or []),
"link": getattr(h, "link", None),
"description": getattr(h, "description", None),
"publisher": getattr(h, "publisher", None),
}
@@ -203,11 +226,83 @@ def _key_str(k: Any) -> str:
return getattr(k, "value", str(k))
def _analysis_to_dict(analysis: Any) -> Optional[Dict[str, Any]]:
"""Compact, JSON-safe snapshot of one TechnicalAnalysis."""
if analysis is None:
return None
indicators = {}
for name, indicator in (
getattr(analysis, "indicators", None) or {}).items():
if str(name).startswith("_"):
continue
signal = getattr(indicator, "signal", None)
indicators[str(name)] = {
"signal": getattr(signal, "name", str(signal)),
"value": _finite(getattr(indicator, "value", None)),
"confidence": _finite(getattr(indicator, "confidence", None)),
"weight": _finite(getattr(indicator, "weight", None)),
"description": getattr(indicator, "description", "") or "",
}
composite = getattr(analysis, "composite_signal", None)
return {
"timestamp": str(getattr(analysis, "timestamp", "")),
"composite_signal": getattr(composite, "name", str(composite)),
"composite_score": _finite(
getattr(analysis, "composite_score", None)),
"confidence": _finite(getattr(analysis, "confidence", None)),
"market_regime": getattr(analysis, "market_regime", None),
"volatility_regime": getattr(analysis, "volatility_regime", None),
"key_levels": {
str(name): _finite(value)
for name, value in (
getattr(analysis, "key_levels", None) or {}).items()
},
"pattern_signals": list(
getattr(analysis, "pattern_signals", None) or []),
"indicators": indicators,
}
def _report_indicators_json(report: Any) -> Optional[str]:
analyses = [getattr(report, "primary_analysis", None)]
analyses.extend(getattr(report, "confirmation_analyses", None) or [])
snapshot = {}
for analysis in analyses:
if analysis is None:
continue
timeframe = str(getattr(analysis, "timeframe", ""))
if timeframe:
snapshot[timeframe] = _analysis_to_dict(analysis)
return _dumps(snapshot) if snapshot else None
def _candles_json(dataframes: Any, limit: int = 14) -> Optional[str]:
"""Last closed OHLCV bars per fetched timeframe, oldest first."""
snapshot = {}
for timeframe, df in (dataframes or {}).items():
if df is None or getattr(df, "empty", True):
continue
bars = []
for timestamp, row in df.tail(limit).iterrows():
bars.append({
"timestamp": _iso(timestamp),
"open": _finite(row.get("Open")),
"high": _finite(row.get("High")),
"low": _finite(row.get("Low")),
"close": _finite(row.get("Close")),
"volume": _finite(row.get("Volume")),
})
if bars:
snapshot[_key_str(timeframe)] = bars
return _dumps(snapshot) if snapshot else None
def _report_row(
report: Any,
context: Any,
source: str,
timeframe: str,
dataframes: Any = None,
) -> tuple:
mtf_signals = getattr(report, "mtf_signals", None) or {}
mtf_json = _dumps({
@@ -271,6 +366,8 @@ def _report_row(
getattr(report, "news_summary", "") or "",
events_json,
headlines_json,
_report_indicators_json(report),
_candles_json(dataframes),
getattr(report, "panel_verdict", None),
_finite(getattr(report, "panel_confidence", None)),
getattr(report, "panel_rationale", None),
@@ -294,30 +391,43 @@ class SignalLogger:
self._lock = threading.Lock()
with self._lock:
self._conn.executescript(_SCHEMA)
self._migrate_panel_columns()
self._migrate_signal_columns()
self._migrate_outlook_columns()
@property
def path(self) -> str:
return self._path
def _migrate_panel_columns(self) -> None:
"""Add panel_* columns to pre-existing tables (idempotent)."""
def _migrate_signal_columns(self) -> None:
"""Add newer signal columns to pre-existing tables (idempotent)."""
existing = {r[1] for r in self._conn.execute("PRAGMA table_info(signals)")}
for name, col_type in _PANEL_COLS:
for name, col_type in _ADDITIVE_SIGNAL_COLS:
if name not in existing:
self._conn.execute(
f"ALTER TABLE signals ADD COLUMN {name} {col_type}")
self._conn.commit()
def _migrate_outlook_columns(self) -> None:
"""Add snapshot columns to pre-existing outlook tables."""
existing = {
r[1] for r in self._conn.execute("PRAGMA table_info(outlooks)")
}
for name, col_type in _ADDITIVE_OUTLOOK_COLS:
if name not in existing:
self._conn.execute(
f"ALTER TABLE outlooks ADD COLUMN {name} {col_type}")
self._conn.commit()
def log(
self,
report: Any,
context: Any = None,
source: str = "live",
timeframe: str = "",
dataframes: Any = None,
) -> int:
"""Insert one scan row; returns the rowid (panel-update key)."""
row = _report_row(report, context, source, timeframe)
row = _report_row(report, context, source, timeframe, dataframes)
with self._lock:
cur = self._conn.execute(_INSERT, row)
self._conn.commit()
@@ -339,8 +449,16 @@ class SignalLogger:
(verdict, confidence, rationale, takes_json, rowid))
self._conn.commit()
def log_outlook(self, outlook: Any, surface: str) -> int:
def log_outlook(self, outlook: Any, surface: str,
payload: Any = None) -> int:
"""Insert one advisory market outlook; returns its rowid."""
payload = payload if isinstance(payload, dict) else {}
technical = {
key: payload.get(key)
for key in ("indicators", "market_regime",
"volatility_regime", "key_levels")
if key in payload
}
row = (
datetime.now(timezone.utc).isoformat(),
surface,
@@ -349,6 +467,9 @@ class SignalLogger:
_finite(getattr(outlook, "confidence", None)),
getattr(outlook, "watch", None),
getattr(outlook, "note", None),
_dumps(payload.get("news")) if "news" in payload else None,
_dumps(technical) if technical else None,
_dumps(payload.get("candles")) if "candles" in payload else None,
)
with self._lock:
cur = self._conn.execute(_OUTLOOK_INSERT, row)
@@ -371,6 +492,7 @@ def log_signal(
context: Any = None,
source: str = "live",
timeframe: str = "",
dataframes: Any = None,
) -> None:
"""Persist one scan if the signal_log config block enables it.
Stashes the rowid on the report (``_signal_log_rowid``) so a post-hoc
@@ -386,7 +508,7 @@ def log_signal(
lg = SignalLogger(path)
_loggers[path] = lg
rowid = lg.log(report, context=context, source=source,
timeframe=timeframe)
timeframe=timeframe, dataframes=dataframes)
try:
report._signal_log_rowid = rowid
except Exception:
@@ -424,7 +546,8 @@ def log_panel_update(report: Any) -> None:
exc_info=True)
def log_outlook(outlook: Any, surface: str = "live") -> None:
def log_outlook(outlook: Any, surface: str = "live",
payload: Any = None) -> None:
"""Persist one advisory outlook when signal logging is enabled.
Outlook persistence is deliberately independent of a signal row: the
@@ -439,7 +562,7 @@ def log_outlook(outlook: Any, surface: str = "live") -> None:
if lg is None:
lg = SignalLogger(path)
_loggers[path] = lg
lg.log_outlook(outlook, surface)
lg.log_outlook(outlook, surface, payload=payload)
except Exception:
logger.warning("signal_log: failed to persist market outlook",
exc_info=True)