Great Trough Detector: exact snapshot, honest boundary¶

import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display
from research.recessionalert.great_trough import (
    load_great_trough, fixture_summary, outcome_matrix, subperiod_outcomes, gap_report, check_causality
)

frame = load_great_trough()
summary = fixture_summary(frame)
pd.Series(summary, name='fixture').to_frame()

rules = pd.DataFrame([
 {'class':'B', 'arm':'GTR < 27.4', 'window':'5 trading days', 'buy':'GTR > 80 after ≥4 days below 80'},
 {'class':'A', 'arm':'GTR < 10.0', 'window':'10 trading days', 'buy':'GTR > 80 after ≥4 days below 80'},
 {'class':'legacy only', 'arm':'GTR < 26 or 27', 'window':'14 or 15 trading days', 'buy':'GTR > 87'},
])
display(rules)
pd.DataFrame({
 'field':['GTR rows','Class A buys','Class B buys','Unique buy dates'],
 'value':[summary['rows'],summary['class_a_signals'],summary['class_b_signals'],summary['unique_signals']]
})

fig, ax = plt.subplots(figsize=(12, 5.5))
ax.plot(frame.index, frame['gtr'], color='#4c72b0', lw=1.0, label='GTR/SIG snapshot')
a = frame[frame['class_a_buy']]; b = frame[frame['class_b_buy']]
ax.scatter(a.index, a['gtr'], marker='^', s=58, color='#2a9d8f', label='Class A buy', zorder=4)
ax.scatter(b.index, b['gtr'], marker='^', s=38, color='#e76f51', label='Class B buy', zorder=3)
for y, label, color in [(10,'A arm 10','#2a9d8f'),(27.4,'B arm 27.4','#e9c46a'),(80,'buy punch 80','#6c757d')]:
    ax.axhline(y, ls='--', lw=.9, color=color, alpha=.8, label=label)
ax.set(title='Pinned Great Trough snapshot · 2015–2026', xlabel='Session', ylabel='Breadth level (0–100)', ylim=(-3,103))
ax.grid(alpha=.2); ax.legend(ncol=3, frameon=False); plt.show()

lag1 = outcome_matrix(frame, execution_lag=1)
rows=[]
for horizon, groups in lag1.items():
    for group in ('combined','class_a','class_b','unconditional'):
        row={'horizon':horizon,'group':group}; row.update(groups[group]); rows.append(row)
outcomes=pd.DataFrame(rows)
outcomes.style.format({'win_rate':'{:.1%}','false_signal_rate':'{:.1%}','mean_return':'{:.2%}','median_return':'{:.2%}'}, na_rep='—')

plot = outcomes[outcomes['group'].isin(['combined','unconditional'])].pivot(index='horizon',columns='group',values='win_rate')
ax=plot.plot.bar(figsize=(9,4.5),color=['#2a9d8f','#9aa0a6'])
ax.set(title='Positive S&P outcome rate after +1-bar entry',xlabel='Trading-day horizon',ylabel='Win rate',ylim=(0,1))
ax.grid(axis='y',alpha=.2); ax.legend(frameon=False); plt.tight_layout(); plt.show()

lag2 = outcome_matrix(frame, execution_lag=2)
stress=[]
for h in lag1:
    for lag, matrix in [(1,lag1),(2,lag2)]:
        r=matrix[h]['combined']; stress.append({'horizon':h,'entry_lag':lag,**r})
display(pd.DataFrame(stress).style.format({'win_rate':'{:.1%}','false_signal_rate':'{:.1%}','mean_return':'{:.2%}','median_return':'{:.2%}'}))
split=[]
for period, groups in subperiod_outcomes(frame).items():
    for group in ('combined','unconditional'):
        split.append({'period':period,'group':group,**groups[group]})
display(pd.DataFrame(split).style.format({'win_rate':'{:.1%}','false_signal_rate':'{:.1%}','mean_return':'{:.2%}','median_return':'{:.2%}'}))
pd.DataFrame([
 {'route':'Local proxy','result':'reject','reason':'EMA(20) 13WK correlation 0.5316; different scale'},
 {'route':'Synthetic extension','result':'reject','reason':'would fit an expressly proprietary smoothing constant'},
 {'route':'Constituent rebuild','result':'reject','reason':'point-in-time NYSE common-stock universe is absent'},
 {'route':'Cached third party','result':'reject','reason':'narrative/charts only; no date ledger'},
 {'route':'Other / OCR','result':'reject','reason':'pixels cannot establish exact signal dates'},
])

display(pd.Series(check_causality(frame), name='causality').to_frame())
pd.Series(gap_report(), name='fail-closed boundary').to_frame()