agent_panel: harden evidence and track outcomes

This commit is contained in:
root
2026-08-23 12:57:05 -04:00
parent 6c5e58f1c3
commit 4dd0ca0b46
7 changed files with 1313 additions and 159 deletions
+384 -19
View File
@@ -1213,6 +1213,7 @@ def _():
assert ts.risk_reward == 2.0
assert ts.entry_price > ts.stop_loss
assert ts.trail_arm_r is None and ts.trail_dist_r is None # unstamped default
assert ts.setup_pattern is None and ts.round_trip_cost is None
@test("trade setups carry trail params from the producing config")
@@ -1228,6 +1229,8 @@ def _():
report.current_price = 3300.0
ts = agg._generate_xau_atr_trade_setup(report, 10.0)
assert ts is not None, "A+ setup with ATR 10 should generate"
assert ts.strategy == "XAU_ATR"
assert ts.round_trip_cost == config.signals.round_trip_cost
assert ts.trail_arm_r == 0.70, "setup must carry the producing config's arm"
assert ts.trail_dist_r == 0.85, "setup must carry the producing config's dist"
@@ -2768,6 +2771,7 @@ def _():
ts = rep.trade_setup
assert ts is not None, "pattern setup not generated"
assert ts.strategy == "XAU_CP" and ts.direction == "BUY"
assert ts.setup_pattern == "double_bottom"
assert ts.stop_loss < ts.entry_price
assert abs((ts.take_profit_1 - ts.entry_price) - 19.0) < 1e-6, \
"TP1 must be the measured move"
@@ -2776,6 +2780,31 @@ def _():
"tag marker must sit at reasons[0]"
@test("CP gate: accepted setup removes stale vetoes and deduplicates extension context")
def _():
cfg = _cp_config()
df = make_cp_df(double_bottom_closes())
rep = _cp_report(df, cfg)
rep.warnings = [
"SPREAD GATE: old core target failed",
"EXHAUSTED TREND FILTER: C-grade blocked in exhausted trend",
"M5 FILTER: no MTF agreement — M5 signals require confirmation",
"FAR FROM EMA200 (4.0%) — over-extended",
"TREND EXHAUSTION — price extended >3.5 ATR past swing low",
"NO PULLBACK — price 3.0 ATR from EMA21",
"Extended — RSI 78 in uptrend",
"EVENT RISK: CPI in 20m",
"NEAR EMA200 (0.3%) — chop risk",
]
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
assert rep.trade_setup is not None
assert rep.warnings == [
"TREND EXHAUSTION — price extended >3.5 ATR past swing low",
"EVENT RISK: CPI in 20m",
"NEAR EMA200 (0.3%) — chop risk",
], rep.warnings
@test("CP gate: MTF opposition blocks the pattern trade")
def _():
cfg = _cp_config()
@@ -2828,8 +2857,11 @@ def _():
cfg.signals.min_tp_cost_multiple = 1000.0 # $300 minimum target
df = make_cp_df(double_bottom_closes())
rep = _cp_report(df, cfg)
rep.warnings = ["SPREAD GATE: prior warning", "NO PULLBACK — context"]
before = list(rep.warnings)
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
assert rep.trade_setup is None
assert rep.warnings == before, "rejected CP must not rewrite diagnostics"
@test("CP pipeline: fires end-to-end when core generators pass")
@@ -6345,6 +6377,7 @@ class _M21cBot:
def _m21c_reset():
_tb21._scan_cooldowns.clear()
_tb21._pending_breach.clear()
_tb21._alert_locks.clear()
_tb21._last_alert_fp.clear()
_tb21._last_alert_ts.clear()
_tb21._last_alert_setup.clear()
@@ -6858,11 +6891,12 @@ def _():
assert pbase._is_forex_weekend(T("2026-08-01 03:00")) is True # Saturday
@test("weekend calendar: all three copies agree hourly across DST transitions")
@test("weekend calendar: all four copies agree hourly across DST transitions")
def _():
import providers.base as pbase
import signal_aggregator as sa_mod
import gpu_backtest as _gb
import advisory_outcome as ao
cfg = PlatformConfig()
cfg.xau_strategy.weekend_filter = True
cfg.xau_strategy.allowed_weekdays = [] # isolate the weekend gate
@@ -6872,7 +6906,9 @@ def _():
a = pbase._is_forex_weekend(ts)
b = sa_mod._is_forex_weekend(ts)
c = _gb._is_xau_blocked_by_calendar(ts, cfg) == "WEEKEND_FILTER"
assert a == b == c, f"{ts}: base={a} aggregator={b} backtest={c}"
d = ao._is_forex_weekend(ts.to_pydatetime())
assert a == b == c == d, \
f"{ts}: base={a} aggregator={b} backtest={c} advisory={d}"
@test("FB2_EMA_LO: garbage value degrades to tuned default, scan survives")
@@ -7077,12 +7113,13 @@ def _panel_report(quality="A", signal=SignalStrength.BUY):
config.primary_timeframe = Timeframe.H1
config.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
agg = SignalAggregator(config)
dfs = {Timeframe.H1: make_ohlcv(300, seed=42),
Timeframe.M15: make_ohlcv(300, seed=43),
Timeframe.H4: make_ohlcv(300, seed=44)}
dfs = {Timeframe.H1: make_xau_ohlcv(300, seed=42),
Timeframe.M15: make_xau_ohlcv(300, seed=43),
Timeframe.H4: make_xau_ohlcv(300, seed=44)}
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
report.final_signal = signal
report.signal_quality = quality
report.trade_setup = None
return report
@@ -7090,19 +7127,32 @@ def _panel_report(quality="A", signal=SignalStrength.BUY):
def _():
report = _panel_report()
now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
directional = [NewsSentiment(
headline=f"h{i}", source="rss", timestamp=now,
currencies_mentioned=["USD"], sentiment_score=0.2, keywords=[])
for i in range(15)]
neutral = NewsSentiment(
headline="neutral", source="rss", timestamp=now,
currencies_mentioned=["XAU"], sentiment_score=0.05, keywords=[])
unrelated = NewsSentiment(
headline="unrelated", source="rss", timestamp=now,
currencies_mentioned=["EUR"], sentiment_score=0.8, keywords=[])
ctx = FundamentalContext(
news_headlines=[NewsSentiment(
headline=f"h{i}", source="rss", timestamp=now,
currencies_mentioned=["USD"], sentiment_score=0.1, keywords=[])
for i in range(15)],
news_headlines=[neutral, unrelated] + directional,
events_today=[EconomicEvent(
time=now, currency="USD", impact="high", event=f"e{i}",
forecast="1.0", actual="2.0") for i in range(12)],
)
macro = ap._macro_payload(report, ctx)
assert len(macro["headlines"]) == 10, "headline cap"
assert all(h["gold_influence"] == "bearish" for h in macro["headlines"])
assert all(h["headline"] not in ("neutral", "unrelated")
for h in macro["headlines"])
assert "sentiment" not in macro, "neutral aggregate sentiment must be omitted"
assert "news_summary" not in macro, "unfiltered summary must not bypass filtering"
assert len(macro["events"]) == 8, "event cap"
assert macro["events"][0]["released"] is True
assert macro["events"][0]["time"] and macro["events"][0]["surprise"] == "positive"
assert ap._macro_payload(report, None)["headlines"] == []
tech = ap._technical_payload(report)
@@ -7113,19 +7163,38 @@ def _():
assert not hidden, f"private/weight-0 leaked: {hidden}"
assert tech["brief"]["direction"] == "BUY"
assert tech["brief"]["timeframe"] == report.primary_analysis.timeframe
assert tech["strategy"]["code"] == "NO_SETUP"
assert "setup" not in ap._risk_payload(report), "no setup when None"
report.trade_setup = TradeSetup(
pair="XAU/USD", display_name="XAU/USD", direction="SELL",
entry_price=2000.0, stop_loss=2010.0, take_profit_1=1985.0,
take_profit_2=1970.0, risk_reward=1.5, risk_pips=100.0,
reward_pips=150.0, pip_value=0.1)
reward_pips=150.0, pip_value=0.1, strategy="XAU_CP",
setup_pattern="double_top", round_trip_cost=0.30,
trail_arm_r=0.6, trail_dist_r=0.85)
tech = ap._technical_payload(report)
assert tech["strategy"]["code"] == "XAU_CP"
assert tech["strategy"]["setup_pattern"] == "double_top"
risk = ap._risk_payload(report)
assert risk["setup"]["risk_reward"] == 1.5
assert risk["setup"]["risk_reward_to_advisory_tp1"] == 1.5
assert risk["setup"]["strategy"]["setup_pattern"] == "double_top"
assert risk["setup"]["trailing"] == {
"enabled": True, "arm_r": 0.6, "distance_r": 0.85,
"tp1_tp2_are_advisory": True}
assert risk["setup"]["round_trip_cost"] == 0.3
assert risk["setup"]["atr"] > 0 and risk["setup"]["stop_distance_atr"] > 0
assert "spread" not in risk, "candle range must not be mislabeled as spread"
assert risk["brief"]["direction"] == "SELL", "setup direction wins"
contra = ap._contrarian_payload(report)
contra = ap._contrarian_payload(report, ctx)
assert len(contra["bullish_reasons"]) <= 3
assert contra["setup"]["strategy"]["code"] == "XAU_CP"
assert contra["market_snapshot"]["basis"] == "latest_closed_candle"
assert contra["market_snapshot"]["bar_timestamp"] == report.timestamp
assert contra["indicators"] == tech["indicators"]
assert len(contra["news"]) == 10
assert "news_summary" not in contra
for payload in (macro, tech, risk, contra):
json.dumps(payload, sort_keys=True, allow_nan=False)
@@ -7180,6 +7249,7 @@ def _():
lc._client = stub
v = ap.run_panel(_panel_report())
assert v.verdict == "OPPOSE" and v.confidence == 72.4
assert v.payload_hash.startswith("sha256:")
assert len(v.rationale) == 240, "rationale truncated"
assert [t.persona for t in v.takes] == \
["macro", "technical", "risk", "contrarian"]
@@ -7216,6 +7286,8 @@ def _():
assert ap.attach_panel(report, _PANEL_ON,
context=FundamentalContext()) is True
assert report.panel_verdict == "AGREE"
assert report.panel_cache_status == "MISS"
assert report.panel_created_at and report.panel_payload_hash.startswith("sha256:")
assert len(stub.calls) == 5
# same bar again → exact-key hit
@@ -7223,6 +7295,9 @@ def _():
assert ap.attach_panel(report2, _PANEL_ON,
context=FundamentalContext()) is True
assert report2.panel_verdict == "AGREE" and len(stub.calls) == 5
assert report2.panel_cache_status == "EXACT_HIT"
assert report2.panel_payload_hash == report.panel_payload_hash
assert report2.panel_created_at == report.panel_created_at
# newer bar, same symbol/tf/direction → reuse-TTL hit
report3 = _panel_report()
@@ -7230,13 +7305,24 @@ def _():
assert ap.attach_panel(report3, _PANEL_ON,
context=FundamentalContext()) is True
assert len(stub.calls) == 5, "persisting signal must not re-spend"
assert report3.panel_cache_status == "TTL_REUSE"
assert report3.panel_payload_hash == report.panel_payload_hash
# Changed evidence must not inherit a direction-only cached verdict.
changed = _panel_report()
changed.timestamp = "2099-01-01 01:05:00+00:00"
changed.current_price += 1.0
assert ap.attach_panel(changed, _PANEL_ON,
context=FundamentalContext()) is True
assert len(stub.calls) == 10
assert changed.panel_cache_status == "MISS"
# TTL 0 → the reuse layer is off; a new bar re-runs the panel
report4 = _panel_report()
report4.timestamp = "2099-01-01 02:00:00+00:00"
assert ap.attach_panel(report4, {**_PANEL_ON, "reuse_ttl_seconds": 0},
context=FundamentalContext()) is True
assert len(stub.calls) == 10
assert len(stub.calls) == 15
finally:
lc._client = None
ap._dp = None
@@ -7271,6 +7357,35 @@ def _():
ap.reset_panel_state()
@test("agent_panel: identical concurrent attaches are single-flight")
def _():
class _SlowPanelStub(_PanelStub):
def chat_json(self, system_prompt, user_payload):
time.sleep(0.03)
return super().chat_json(system_prompt, user_payload)
try:
stub = _SlowPanelStub(dict(_ALL_OK))
lc._client = stub
ap._dp = _NullDP()
reports = [_panel_report(), _panel_report()]
results = []
threads = [threading.Thread(
target=lambda report=r: results.append(ap.attach_panel(
report, _PANEL_ON, context=FundamentalContext())))
for r in reports]
for thread in threads:
thread.start()
for thread in threads:
thread.join(timeout=5)
assert results == [True, True]
assert len(stub.calls) == 5, "one evidence payload must spend one panel"
assert {r.panel_cache_status for r in reports} == {"MISS", "EXACT_HIT"}
finally:
lc._client = None
ap._dp = None
ap.reset_panel_state()
@test("agent_panel: enabled/surface/min_quality gates; force bypasses")
def _():
try:
@@ -7345,6 +7460,9 @@ def _():
assert report.panel_confidence == 0.0
assert report.panel_rationale == ""
assert report.panel_takes == []
assert report.panel_created_at == ""
assert report.panel_cache_status == ""
assert report.panel_payload_hash == ""
assert stub.calls == [], \
"the replayable core must never invoke the panel"
finally:
@@ -7352,6 +7470,31 @@ def _():
ap.reset_panel_state()
@test("agent_panel: attach remains advisory-only")
def _():
try:
lc._client = _PanelStub(dict(_ALL_OK))
ap._dp = _NullDP()
report = _panel_report()
before = (
report.final_signal, report.final_score, report.final_confidence,
report.signal_quality, report.action_recommendation,
report.trade_setup, list(report.warnings),
)
assert ap.attach_panel(report, _PANEL_ON,
context=FundamentalContext()) is True
after = (
report.final_signal, report.final_score, report.final_confidence,
report.signal_quality, report.action_recommendation,
report.trade_setup, list(report.warnings),
)
assert after == before, "panel must mutate only advisory panel fields"
finally:
lc._client = None
ap._dp = None
ap.reset_panel_state()
@test("report_to_dict carries panel keys → dashboard snapshot propagation")
def _():
from main import report_to_dict as _rtd
@@ -7482,6 +7625,11 @@ def _():
ap._CONTRARIAN_SYSTEM):
assert "candles" not in by_prompt[sys_p], \
"candles are technical-persona evidence only"
contra = by_prompt[ap._CONTRARIAN_SYSTEM]
assert contra["setup"] is None
assert contra["market_snapshot"]["basis"] == "latest_closed_candle"
assert contra["indicators"]
assert "news" in contra
finally:
lc._client = None
ap._dp = None
@@ -7494,6 +7642,11 @@ def _():
assert "max 15 words" in ap._TAKE_CONTRACT
assert "max 30 words" in ap._SYNTH_SYSTEM
assert "ATR units" in ap._TECH_SYSTEM
assert "setup_pattern" in ap._TECH_SYSTEM
assert "tp1_tp2_are_advisory" not in ap._RISK_SYSTEM # prose, not wire key
assert "reference levels" in ap._RISK_SYSTEM
assert "gold_influence" in ap._MACRO_SYSTEM
assert "market_snapshot" in ap._CONTRARIAN_SYSTEM
for prompt in (ap._MACRO_SYSTEM, ap._TECH_SYSTEM, ap._RISK_SYSTEM,
ap._CONTRARIAN_SYSTEM):
assert ap._TAKE_CONTRACT in prompt, \
@@ -7537,6 +7690,7 @@ def _make_log_report():
entry_price=2450.50, stop_loss=2445.00, take_profit_1=2458.00,
take_profit_2=2468.00, risk_reward=1.6, risk_pips=5.5,
reward_pips=8.8, pip_value=1.0, strategy="XAU_PB",
setup_pattern=None, round_trip_cost=0.30,
trail_arm_r=0.70, trail_dist_r=0.85,
)
report.primary_analysis = TechnicalAnalysis(
@@ -7625,6 +7779,8 @@ def _():
assert mtf["scores"][Timeframe.M5.value] == 0.6
setup = json.loads(row["setup_json"])
assert setup["strategy"] == "XAU_PB"
assert setup["setup_pattern"] is None
assert setup["round_trip_cost"] == 0.30
assert setup["entry"] == 2450.50 and setup["trail_arm_r"] == 0.70
reasons = json.loads(row["reasons_json"])
assert "MTF agree" in reasons["bullish"]
@@ -7670,21 +7826,32 @@ def _():
report.panel_verdict = "OPPOSE"
report.panel_confidence = 64.0
report.panel_rationale = "risk too thin"
report.panel_created_at = "2026-08-01T12:00:03+00:00"
report.panel_cache_status = "MISS"
report.panel_payload_hash = "sha256:" + "a" * 64
report.panel_takes = [{"persona": "risk", "stance": "oppose",
"confidence": 70.0, "note": "n"}]
lg.update_panel(rowid, report.panel_verdict, 64.0,
report.panel_rationale,
slog._dumps(report.panel_takes))
slog._dumps(report.panel_takes),
created_at=report.panel_created_at,
cache_status=report.panel_cache_status,
payload_hash=report.panel_payload_hash)
lg.close()
conn = sqlite3.connect(path)
row = conn.execute(
"SELECT panel_verdict, panel_confidence, panel_rationale,"
" panel_takes_json FROM signals WHERE id=?",
" panel_takes_json, panel_created_at, panel_cache_status,"
" panel_payload_hash, advisory_outcome_status"
" FROM signals WHERE id=?",
(rowid,)).fetchone()
conn.close()
assert row[0] == "OPPOSE" and row[1] == 64.0
assert row[2] == "risk too thin"
assert json.loads(row[3])[0]["persona"] == "risk"
assert row[4] == report.panel_created_at and row[5] == "MISS"
assert row[6] == report.panel_payload_hash
assert row[7] == "PENDING"
finally:
slog.reset_loggers_for_tests()
os.remove(path)
@@ -7721,15 +7888,21 @@ def _():
report.panel_confidence = 81.0
report.panel_rationale = "all aligned"
report.panel_takes = []
report.panel_created_at = "2026-08-01T12:00:05+00:00"
report.panel_cache_status = "EXACT_HIT"
report.panel_payload_hash = "sha256:" + "b" * 64
slog.log_panel_update(report)
conn = sqlite3.connect(path)
row = conn.execute(
"SELECT panel_verdict, panel_confidence, panel_rationale,"
" panel_takes_json FROM signals").fetchone()
" panel_takes_json, panel_created_at, panel_cache_status,"
" panel_payload_hash, advisory_outcome_status FROM signals").fetchone()
conn.close()
assert row[0] == "AGREE" and row[1] == 81.0
assert row[2] == "all aligned"
assert json.loads(row[3]) == []
assert row[4] == report.panel_created_at and row[5] == "EXACT_HIT"
assert row[6] == report.panel_payload_hash and row[7] == "PENDING"
# disabled → no-op, never raises
os.environ["SIGNAL_LOG_ENABLED"] = "false"
@@ -7765,6 +7938,174 @@ def _():
os.remove(path)
@test("advisory outcome: next-open trailing ignores TP1 and ratchets at close")
def _():
from advisory_outcome import advance_outcome, initial_outcome
setup = {
"direction": "BUY", "entry": 100.0, "stop_loss": 90.0,
"take_profit_1": 105.0, "strategy": "GENERIC",
"setup_pattern": None, "trail_arm_r": 0.6,
"trail_dist_r": 0.75,
}
indicators = {"5min": {"timestamp": "2026-08-03T10:00:00+00:00"}}
status, state = initial_outcome(setup, indicators, "M5")
assert status == "PENDING"
state = advance_outcome(state, [{
"timestamp": "2026-08-03T10:00:00+00:00",
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}, {
"timestamp": "2026-08-03T10:05:00+00:00",
"open": 101.0, "high": 112.0, "low": 100.0, "close": 110.0,
}])
assert state["status"] == "PENDING", "advisory TP1 is not a hard exit"
assert state["progress"]["entry_price"] == 101.0
assert state["progress"]["initial_stop"] == 91.0
assert state["progress"]["current_stop"] == 104.5
state = advance_outcome(state, [{
"timestamp": "2026-08-03T10:05:00+00:00",
"open": 101.0, "high": 112.0, "low": 100.0, "close": 110.0,
}, {
"timestamp": "2026-08-03T10:10:00+00:00",
"open": 108.0, "high": 109.0, "low": 104.0, "close": 105.0,
}])
assert state["status"] == "RESOLVED"
result = state["result"]
assert result["exit_reason"] == "TRAIL" and result["bars_held"] == 2
assert abs(result["net_r_065"] - 0.285) < 1e-12
_, broken = initial_outcome(setup, indicators, "M5")
broken = advance_outcome(broken, [{
"timestamp": "2026-08-03T10:10:00+00:00",
"open": 108.0, "high": 109.0, "low": 104.0, "close": 105.0,
}])
assert broken["status"] == "UNRESOLVABLE"
assert "continuity" in broken["reason"]
_, gapped = initial_outcome(setup, indicators, "M5")
gapped = advance_outcome(gapped, [{
"timestamp": "2026-08-03T10:00:00+00:00",
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}, {
"timestamp": "2026-08-03T10:10:00+00:00",
"open": 101.0, "high": 102.0, "low": 100.0, "close": 101.0,
}])
assert gapped["status"] == "UNRESOLVABLE"
assert "internal gap" in gapped["reason"]
weekend_indicators = {
"5min": {"timestamp": "2026-08-07T20:55:00+00:00"}}
_, weekend = initial_outcome(setup, weekend_indicators, "M5")
weekend = advance_outcome(weekend, [{
"timestamp": "2026-08-07T20:55:00+00:00",
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}, {
"timestamp": "2026-08-09T22:00:00+00:00",
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}])
assert weekend["status"] == "PENDING", \
"the scheduled FX weekend closure is not a data gap"
maintenance_indicators = {
"5min": {"timestamp": "2026-08-03T20:55:00+00:00"}}
_, maintenance = initial_outcome(setup, maintenance_indicators, "M5")
maintenance = advance_outcome(maintenance, [{
"timestamp": "2026-08-03T20:55:00+00:00",
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}, {
"timestamp": "2026-08-03T22:00:00+00:00",
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}])
assert maintenance["status"] == "PENDING", \
"Oanda's scheduled daily XAU closure is not a data gap"
for timeframe, normalized, signal_at, next_at in (
("M30", "30min", "2026-08-03T10:00:00+00:00",
"2026-08-03T10:30:00+00:00"),
("D1", "1day", "2026-08-03T00:00:00+00:00",
"2026-08-04T00:00:00+00:00"),
):
stored_key = "1d" if timeframe == "D1" else normalized
alias_indicators = {stored_key: {"timestamp": signal_at}}
_, alias_state = initial_outcome(setup, alias_indicators, timeframe)
alias_state = advance_outcome(alias_state, [{
"timestamp": signal_at,
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}, {
"timestamp": next_at,
"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.0,
}])
assert alias_state["status"] == "PENDING", timeframe
@test("signal_log: pending advisory outcome resolves from later snapshots")
def _():
import sqlite3
import signal_log as slog
fd, path = tempfile.mkstemp(suffix=".db")
os.close(fd)
try:
lg = slog.SignalLogger(path)
report = _make_log_report()
rowid = lg.log(report, source="test", timeframe="M5")
lg.update_panel(
rowid, "OPPOSE", 70.0, "test", "[]",
created_at="2026-08-01T12:00:03+00:00",
cache_status="MISS", payload_hash="sha256:" + "c" * 64,
)
bars = slog._dumps({"5min": [{
"timestamp": "2026-08-01T12:00:00+00:00",
"open": 2450.0, "high": 2451.0, "low": 2449.0,
"close": 2450.5, "volume": 1.0,
}, {
"timestamp": "2026-08-01T12:05:00+00:00",
"open": 2450.5, "high": 2451.0, "low": 2444.0,
"close": 2445.0, "volume": 1.0,
}]})
assert lg.resolve_pending("XAU/USD", bars) == 1
row = lg._conn.execute(
"SELECT advisory_outcome_status, advisory_r_multiple,"
" advisory_outcome_json FROM signals WHERE id=?", (rowid,)
).fetchone()
assert row[0] == "RESOLVED" and row[1] < -1.0
outcome = json.loads(row[2])
assert outcome["result"]["exit_reason"] == "SL"
assert outcome["model"] == "stored-closed-mid-next-open-trail-v1"
# Overlapping snapshots are idempotent after resolution.
assert lg.resolve_pending("XAU/USD", bars) == 0
d1_report = _make_log_report()
d1_report.timestamp = "2026-08-03T00:00:00+00:00"
d1_report.primary_analysis.timeframe = Timeframe.D1.value
d1_report.primary_analysis.timestamp = d1_report.timestamp
d1_rowid = lg.log(d1_report, source="test", timeframe="D1")
lg.update_panel(
d1_rowid, "AGREE", 70.0, "test", "[]",
created_at="2026-08-03T00:00:03+00:00",
cache_status="MISS", payload_hash="sha256:" + "d" * 64,
)
d1_bars = slog._dumps({"1d": [{
"timestamp": "2026-08-03T00:00:00+00:00",
"open": 2450.0, "high": 2451.0, "low": 2449.0,
"close": 2450.5, "volume": 1.0,
}, {
"timestamp": "2026-08-04T00:00:00+00:00",
"open": 2450.5, "high": 2451.0, "low": 2449.0,
"close": 2450.0, "volume": 1.0,
}]})
assert lg.resolve_pending("XAU/USD", d1_bars) == 1
d1_state = json.loads(lg._conn.execute(
"SELECT advisory_outcome_json FROM signals WHERE id=?",
(d1_rowid,)).fetchone()[0])
assert d1_state["status"] == "PENDING"
assert d1_state["progress"]["entry_at"] == \
"2026-08-04T00:00:00+00:00"
lg.close()
finally:
slog.reset_loggers_for_tests()
if os.path.exists(path):
os.remove(path)
@test("signal_log: module entry honors SIGNAL_LOG_ENABLED/PATH env")
def _():
import signal_log as slog
@@ -8025,11 +8366,34 @@ def _():
" watch TEXT, note TEXT)")
con.commit()
con.close()
logger = slog.SignalLogger(path)
barrier = threading.Barrier(4)
loggers = []
errors = []
def open_logger():
try:
barrier.wait()
loggers.append(slog.SignalLogger(path))
except Exception as exc:
errors.append(exc)
threads = [threading.Thread(target=open_logger) for _ in range(4)]
for thread in threads:
thread.start()
for thread in threads:
thread.join(timeout=5)
assert not errors, f"concurrent additive migration failed: {errors}"
assert len(loggers) == 4
logger = loggers[0]
columns = {
row[1] for row in logger._conn.execute("PRAGMA table_info(signals)")
}
assert {"indicators_json", "candles_json"} <= columns
assert {
"indicators_json", "candles_json", "panel_created_at",
"panel_cache_status", "panel_payload_hash",
"advisory_outcome_status", "advisory_r_multiple",
"advisory_outcome_json",
} <= columns
outlook_columns = {
row[1] for row in logger._conn.execute("PRAGMA table_info(outlooks)")
}
@@ -8048,7 +8412,8 @@ def _():
"candles": {"1h": {"bars": [[-0.2, 0.1, -0.3, 0.0]]}},
}
rowid = logger.log_outlook(result, surface="test", payload=payload)
logger.close()
for open_instance in loggers:
open_instance.close()
con = sqlite3.connect(path)
row = con.execute(
"SELECT surface, bias, setup_forming, confidence, watch, note, "