#!/usr/bin/env python3 """ Builds retrospective.json: a daily hits/misses/missed-opportunities summary for the dashboard's Retrospective panel. Hits/misses come from realized P/L on positions.json's closed_positions — real fills only, never estimated. Missed opportunities are new_buys candidates from daily_snapshots/ that were never entered, checked against a live TMX quote to see if they ran without us. Run manually: python3 generate_retrospective.py Cron (once daily, after TSX close, MDT): 5 14 * * 1-5 /usr/bin/python3 /media/raid/rshare/SwingTrader/generate_retrospective.py >> /media/raid/rshare/SwingTrader/daily_run.log 2>&1 """ import json import re import sys from datetime import date, timedelta from pathlib import Path BASE = Path(__file__).parent sys.path.insert(0, str(BASE)) from is_trading_day import is_trading_day from tmx_quotes import get_quote if not is_trading_day(): print(f'[generate_retrospective] {date.today()} is not a TSX trading day -- skipping') sys.exit(0) POSITIONS_FILE = BASE / 'positions.json' SNAPSHOTS_DIR = BASE / 'daily_snapshots' CRITERIA_FILE = BASE / 'screen_criteria.json' OUTPUT_FILE = BASE / 'retrospective.json' # daily_snapshots history was cleared on 2026-08-01; nothing before that exists to mine. CANDIDATE_TRACKING_START = '2026-08-01' # A passed-on candidate must clear this move before it counts as "missed" (real # noise/slippage below this isn't a meaningful missed opportunity). MISSED_OPP_THRESHOLD_PCT = 5.0 # Too soon after the scan to judge the candidate either way. MIN_DAYS_BEFORE_JUDGING = 3 # If we entered the same ticker within this many days of a scan flagging it, # treat it as "acted on" (belongs in hits/misses, not missed opportunities). ENTRY_WINDOW_DAYS = 10 # Distinct tickers actually traded (open + closed) since the program started. # Audit of daily_run.log back to April found only 7 -- CNQ, K, AGI, ENB, SU, BNS, # QSR -- despite 81 distinct tickers being mentioned across scans. The bottleneck # was conversion (watchlist candidates never resolved), not discovery. NARROW_FUNNEL_TICKER_THRESHOLD = 15 # Don't turn a review_flags pattern into a hard rule off a tiny sample. MIN_SAMPLE_FOR_PATTERN_RULE = 3 # Marker string in an `excluded` entry's `reason` that identifies it as a # recurring-mega-cap deferral (daily_scan_prompt.md Step 2 / screen_criteria.json # funnel_breadth.recurring_megacap_handling). These entries carry a `price_cad` # captured at deferral time (added 2026-08-10) specifically so this script can # check whether a hand-waved name moved without anyone noticing. RECURRING_MEGACAP_TAG = 'funnel_breadth.recurring_megacap_handling' DEFERRAL_TRACKING_START = '2026-08-10' # Which of entry_setups' 5 scoring criteria a watchlist/excluded entry's # waiting_for/reason text cites as the one still-unconfirmed point, keyed by # a regex over that text. Added 2026-08-12 after an audit found 20 of 31 # "close but not quite" entries since 2026-08-01 were stuck specifically on # volume (e.g. GIL carried 3 straight scans, 08-07 through 08-11, because no # source stated a precomputed "volume vs 20-day average" ratio) -- see # daily_scan_prompt.md's Step 3 search-format fix on the same date, which now # explicitly asks for current + average volume so the ratio can be computed # directly instead of requiring a source to state it outright. UNCONFIRMED_CRITERION_PATTERNS = { 'volume': r'volume', 'resistance_level': r'resistance', 'candle_pattern': r'candle', 'macd_reading': r'macd.{0,20}(not found|conflict|unconfirmed)', 'ema_proximity': r'within.{0,15}%.{0,15}ema|proximity.{0,15}ema', } def load_json(path, default=None): if not path.exists(): return default with open(path) as f: return json.load(f) def realized_pl(pos): exit_price = pos.get('exit_price_cad') entry_price = pos.get('entry_price_cad') shares = pos.get('shares') if exit_price is None or entry_price is None or shares is None: return None return { 'pl_dollars': round((exit_price - entry_price) * shares, 2), 'pl_pct': round((exit_price - entry_price) / entry_price * 100, 2), } def build_hits_misses(positions): hits, misses, incomplete = [], [], [] for p in positions.get('closed_positions', []): if p.get('_example'): continue entry = { 'ticker': p.get('ticker'), 'company': p.get('company', ''), 'entry_date': p.get('entry_date'), 'close_date': p.get('close_date'), 'entry_price_cad': p.get('entry_price_cad'), 'exit_price_cad': p.get('exit_price_cad'), 'shares': p.get('shares'), 'setup_type': p.get('setup_type', ''), 'close_reason': p.get('close_reason', ''), } pl = realized_pl(p) if pl is None: incomplete.append(entry) continue entry.update(pl) (hits if pl['pl_dollars'] > 0 else misses).append(entry) hits.sort(key=lambda x: x['pl_dollars'], reverse=True) misses.sort(key=lambda x: x['pl_dollars']) return hits, misses, incomplete def already_acted_on(ticker, scan_date_str, positions): try: scan_dt = date.fromisoformat(scan_date_str) except ValueError: return False window_end = scan_dt + timedelta(days=ENTRY_WINDOW_DAYS) all_positions = positions.get('open_positions', []) + positions.get('closed_positions', []) for p in all_positions: if (p.get('ticker') or '').upper() != ticker.upper(): continue ed = p.get('entry_date') if not ed: continue try: entry_dt = date.fromisoformat(ed) except ValueError: continue if scan_dt <= entry_dt <= window_end: return True return False def build_missed_opportunities(positions): missed, passed_ok, pending = [], [], [] if not SNAPSHOTS_DIR.exists(): return missed, passed_ok, pending seen = set() today = date.today() for snap_path in sorted(SNAPSHOTS_DIR.glob('swing_trade_*.json')): snap = load_json(snap_path, {}) or {} scan_date = snap.get('scan_date') if not scan_date or scan_date < CANDIDATE_TRACKING_START: continue for c in (snap.get('signals', {}) or {}).get('new_buys', []) or []: ticker = c.get('ticker') ref_price = c.get('price_cad') if not ticker or ref_price is None: continue key = (ticker.upper(), scan_date) if key in seen: continue seen.add(key) if already_acted_on(ticker, scan_date, positions): continue # we took the trade -- lives in hits/misses instead record = { 'ticker': ticker, 'company': c.get('company', ''), 'scan_date': scan_date, 'price_at_scan_cad': ref_price, 'setup_type': c.get('setup_type', ''), 'setup_score': c.get('setup_score'), } try: scan_dt = date.fromisoformat(scan_date) except ValueError: continue days_since = (today - scan_dt).days quote = get_quote(ticker) if quote is None or quote.get('price') is None: record['status'] = 'price_unavailable' pending.append(record) continue current_price = quote['price'] move_pct = round((current_price - ref_price) / ref_price * 100, 2) record['current_price_cad'] = current_price record['move_pct'] = move_pct if days_since < MIN_DAYS_BEFORE_JUDGING: record['status'] = 'too_early' pending.append(record) elif move_pct >= MISSED_OPP_THRESHOLD_PCT: record['status'] = 'missed' missed.append(record) else: record['status'] = 'passed_correctly' passed_ok.append(record) missed.sort(key=lambda x: x['move_pct'], reverse=True) return missed, passed_ok, pending def build_deferred_megacap_review(positions): """Checks excluded entries tagged recurring_megacap_handling for a move. Those entries are hand-waved past without full scoring to save research budget (see screen_criteria.json funnel_breadth.recurring_megacap_handling) -- this is the same shape as build_missed_opportunities, but starting from a deferral instead of a scored-and-passed new_buys candidate, to catch the case where the deferral itself cost something. """ missed, passed_ok, pending = [], [], [] if not SNAPSHOTS_DIR.exists(): return missed, passed_ok, pending seen = set() today = date.today() for snap_path in sorted(SNAPSHOTS_DIR.glob('swing_trade_*.json')): snap = load_json(snap_path, {}) or {} scan_date = snap.get('scan_date') if not scan_date or scan_date < DEFERRAL_TRACKING_START: continue for c in snap.get('excluded', []) or []: ticker = c.get('ticker') reason = c.get('reason') or '' ref_price = c.get('price_cad') if not ticker or RECURRING_MEGACAP_TAG not in reason or ref_price is None: continue key = (ticker.upper(), scan_date) if key in seen: continue seen.add(key) if already_acted_on(ticker, scan_date, positions): continue # we took the trade -- lives in hits/misses instead record = { 'ticker': ticker, 'scan_date': scan_date, 'price_at_deferral_cad': ref_price, } try: scan_dt = date.fromisoformat(scan_date) except ValueError: continue days_since = (today - scan_dt).days quote = get_quote(ticker) if quote is None or quote.get('price') is None: record['status'] = 'price_unavailable' pending.append(record) continue current_price = quote['price'] move_pct = round((current_price - ref_price) / ref_price * 100, 2) record['current_price_cad'] = current_price record['move_pct'] = move_pct if days_since < MIN_DAYS_BEFORE_JUDGING: record['status'] = 'too_early' pending.append(record) elif move_pct >= MISSED_OPP_THRESHOLD_PCT: record['status'] = 'missed' missed.append(record) else: record['status'] = 'passed_correctly' passed_ok.append(record) missed.sort(key=lambda x: x['move_pct'], reverse=True) return missed, passed_ok, pending def distinct_traded_tickers(positions): tickers = set() for p in positions.get('open_positions', []) + positions.get('closed_positions', []): if p.get('ticker') and not p.get('_example'): tickers.add(p['ticker'].upper()) return tickers def build_unenriched_position_lessons(positions, grace_days): lessons = [] today = date.today() for p in positions.get('open_positions', []): if p.get('_example') or not p.get('ticker'): continue try: entry_dt = date.fromisoformat(p['entry_date']) if p.get('entry_date') else None except ValueError: entry_dt = None days_held = (today - entry_dt).days if entry_dt else None missing = [f for f in ('stop_loss_cad', 'target_1_cad') if p.get(f) is None] if missing and (days_held is None or days_held >= grace_days): lessons.append({ 'code': 'unenriched_position', 'severity': 'high', 'ticker': p['ticker'], 'days_held': days_held, 'summary': ( f"{p['ticker']} is missing {', '.join(missing)} " f"after {days_held if days_held is not None else 'an unknown number of'} days held -- " "exit/risk signals can't be computed accurately." ), }) return lessons def build_overdue_exit_lessons(positions, latest_snapshot): lessons = [] if not latest_snapshot: return lessons open_by_ticker = { p['ticker'].upper(): p for p in positions.get('open_positions', []) if p.get('ticker') and not p.get('_example') } for e in (latest_snapshot.get('signals', {}) or {}).get('exits', []) or []: ticker = (e.get('ticker') or '').upper() pos = open_by_ticker.get(ticker) if pos is None: continue # already closed -- the recommendation was followed try: entry_dt = date.fromisoformat(pos['entry_date']) if pos.get('entry_date') else None except ValueError: entry_dt = None days_held = (date.today() - entry_dt).days if entry_dt else None lessons.append({ 'code': 'overdue_exit_not_executed', 'severity': 'critical', 'ticker': ticker, 'days_held': days_held, 'scan_date': latest_snapshot.get('scan_date'), 'summary': ( f"{ticker} was flagged EXIT in the most recent scan ({latest_snapshot.get('scan_date')}) " f"and is still open ({days_held if days_held is not None else '?'} days held). " "The recommendation has not been acted on." ), }) return lessons def build_review_flag_patterns(positions): tally = {} for p in positions.get('closed_positions', []): if p.get('_example'): continue pl = realized_pl(p) for flag, val in (p.get('review_flags') or {}).items(): if val is not True: continue bucket = tally.setdefault(flag, {'count': 0, 'losses': 0, 'gains': 0, 'tickers': []}) bucket['count'] += 1 bucket['tickers'].append(p.get('ticker')) if pl is not None: if pl['pl_dollars'] > 0: bucket['gains'] += 1 elif pl['pl_dollars'] < 0: bucket['losses'] += 1 patterns = [] for flag, stats in tally.items(): patterns.append({ 'code': f'pattern_{flag}', 'flag': flag, 'count': stats['count'], 'losses': stats['losses'], 'gains': stats['gains'], 'tickers': stats['tickers'], 'actionable': stats['count'] >= MIN_SAMPLE_FOR_PATTERN_RULE, }) return patterns def build_unconfirmed_criteria_patterns(): """Tallies which entry_setups scoring criterion recent watchlist/excluded entries most often cite as the one confirmation still missing. Scans every daily_snapshots/ file from CANDIDATE_TRACKING_START onward -- not just the latest one -- because the point is to catch a *systemic* research-tooling gap (a criterion sources rarely state outright, like a precomputed volume ratio), which only shows up across many scans, not a single day's watchlist. See UNCONFIRMED_CRITERION_PATTERNS' docstring. """ texts = [] if SNAPSHOTS_DIR.exists(): for snap_path in sorted(SNAPSHOTS_DIR.glob('swing_trade_*.json')): snap = load_json(snap_path, {}) or {} scan_date = snap.get('scan_date') if not scan_date or scan_date < CANDIDATE_TRACKING_START: continue for w in (snap.get('signals', {}) or {}).get('watchlist', []) or []: texts.append(w.get('waiting_for') or '') for e in snap.get('excluded', []) or []: reason = e.get('reason') or '' if '/5' in reason or 'missing' in reason.lower(): texts.append(reason) total = len(texts) tally = {label: 0 for label in UNCONFIRMED_CRITERION_PATTERNS} for t in texts: tl = t.lower() for label, pattern in UNCONFIRMED_CRITERION_PATTERNS.items(): if re.search(pattern, tl): tally[label] += 1 patterns = [] for label, count in tally.items(): if count == 0: continue patterns.append({ 'code': f'unconfirmed_{label}', 'criterion': label, 'count': count, 'total_scored_entries': total, 'pct': round(count / total * 100, 1) if total else None, 'actionable': count >= MIN_SAMPLE_FOR_PATTERN_RULE, }) patterns.sort(key=lambda p: p['count'], reverse=True) return patterns def build_sector_concentration_lessons(positions, criteria): """Code-level check, not a prompt note the LLM is trusted to remember. Always flags 2+ open positions sharing a sector, with severity keyed off active_mode -- aggressive mode explicitly allows concentration in the day's leading sector (screen_criteria.json aggressive_overrides.trade_management .sector_concentration), but this script has no way to know which sector was "the leading one" on entry day, so it surfaces every concentration as real, counted evidence and leaves the "is this the intentional play" call to the scan (which has today's macro context) rather than silently assuming either way. """ active_mode = criteria.get('active_mode', 'conservative') by_sector = {} for p in positions.get('open_positions', []): sector = (p.get('sector') or '').strip() if not sector or p.get('_example'): continue by_sector.setdefault(sector, []).append(p.get('ticker')) lessons = [] for sector, tickers in by_sector.items(): if len(tickers) < 2: continue severity = 'info' if active_mode == 'aggressive' else 'high' lessons.append({ 'code': 'sector_concentration', 'severity': severity, 'sector': sector, 'tickers': tickers, 'active_mode': active_mode, 'summary': ( f'{len(tickers)} open positions are concentrated in {sector} ' f'({", ".join(tickers)}). ' + ( 'Aggressive mode allows this if it is the day\'s intentional ' 'leading-sector play, but still counts toward max_simultaneous_positions -- ' 'confirm that\'s actually why before adding another.' if active_mode == 'aggressive' else 'Correlated risk: a single sector move could hit multiple stops at once.' ) ), }) return lessons def latest_snapshot_file(): if not SNAPSHOTS_DIR.exists(): return None files = sorted(SNAPSHOTS_DIR.glob('swing_trade_*.json')) return files[-1] if files else None def build_lessons(positions, deferred_missed=None): snap_path = latest_snapshot_file() latest_snap = load_json(snap_path, None) if snap_path else None criteria = load_json(CRITERIA_FILE, {}) or {} grace_days = criteria.get('trade_management', {}).get('unenriched_position_grace_days', 5) deferred_missed = deferred_missed or [] unenriched = build_unenriched_position_lessons(positions, grace_days) overdue_exits = build_overdue_exit_lessons(positions, latest_snap) patterns = build_review_flag_patterns(positions) concentration = build_sector_concentration_lessons(positions, criteria) unconfirmed = build_unconfirmed_criteria_patterns() traded = distinct_traded_tickers(positions) narrow_funnel = len(traded) < NARROW_FUNNEL_TICKER_THRESHOLD directives = [] for l in overdue_exits: directives.append( l['summary'] + ' Repeat the EXIT call today with escalated urgency and ' 'state the cumulative days ignored (see daily_scan_prompt.md Step 1a/5).' ) for l in unenriched: directives.append( l['summary'] + ' Prioritize enriching it in Step 1b before continuing.' ) for l in concentration: if l['severity'] == 'high': directives.append( l['summary'] + ' Do not add another correlated position in this sector today.' ) if narrow_funnel: directives.append( f"Only {len(traded)} distinct ticker(s) have been traded across the whole program " f"({', '.join(sorted(traded))}), below the {NARROW_FUNNEL_TICKER_THRESHOLD}-ticker " "breadth target. Apply screen_criteria.json's funnel_breadth rules today: use every " "screener_sources entry, resolve stale watchlist items first, and hit " "min_candidates_evaluated_per_scan / min_sectors_represented before finalizing output." ) for p in patterns: if p['actionable']: directives.append( f"Pattern: {p['count']} closed trades were flagged '{p['flag']}' " f"({p['losses']} losses / {p['gains']} gains -- tickers: {', '.join(p['tickers'])}). " "Factor this into today's entry/risk decisions." ) for u in unconfirmed: if u['actionable'] and u['criterion'] == 'volume': directives.append( f"Pattern: {u['count']} of {u['total_scored_entries']} recent watchlist/excluded scoring " f"notes ({u['pct']}%) cite an unconfirmed volume figure as the one blocking confirmation " "for promotion to new_buys. daily_scan_prompt.md's Step 3 confirmation search was updated " "2026-08-12 to explicitly request current + average volume so the ratio can be computed " "directly instead of requiring a source to state it pre-computed -- make sure today's " "scoring actually uses that, rather than deferring on volume again." ) elif u['actionable']: directives.append( f"Pattern: {u['count']} of {u['total_scored_entries']} recent watchlist/excluded scoring " f"notes ({u['pct']}%) cite an unconfirmed '{u['criterion']}' as the blocking confirmation. " "Worth a closer look if this keeps recurring." ) for d in deferred_missed: directives.append( f"{d['ticker']} was deferred without reconfirmation under " f"funnel_breadth.recurring_megacap_handling on {d['scan_date']} and has since moved " f"{d['move_pct']:+.1f}% (${d['price_at_deferral_cad']} -> ${d['current_price_cad']}). " f"Force a full Step 3/4 reconfirmation on {d['ticker']} today, regardless of day-of-week." ) lessons = { 'unenriched_positions': unenriched, 'overdue_exits': overdue_exits, 'review_flag_patterns': patterns, 'sector_concentration': concentration, 'unconfirmed_criteria_patterns': unconfirmed, 'deferred_megacap_moved': deferred_missed, 'distinct_tickers_traded': sorted(traded), 'distinct_ticker_count': len(traded), 'narrow_funnel_flag': narrow_funnel, 'narrow_funnel_threshold': NARROW_FUNNEL_TICKER_THRESHOLD, } return lessons, directives def main(): positions = load_json(POSITIONS_FILE, {}) or {} hits, misses, incomplete = build_hits_misses(positions) missed, passed_ok, pending = build_missed_opportunities(positions) deferred_missed, deferred_ok, deferred_pending = build_deferred_megacap_review(positions) lessons, lessons_for_next_scan = build_lessons(positions, deferred_missed) realized_total = round( sum(h['pl_dollars'] for h in hits) + sum(m['pl_dollars'] for m in misses), 2 ) n_closed = len(hits) + len(misses) win_rate = round(len(hits) / n_closed * 100, 1) if n_closed else None output = { 'generated_at': date.today().isoformat(), 'summary': { 'realized_pl_cad': realized_total, 'win_rate_pct': win_rate, 'hit_count': len(hits), 'miss_count': len(misses), 'incomplete_count': len(incomplete), 'missed_opportunity_count': len(missed), 'deferred_megacap_missed_count': len(deferred_missed), }, 'hits': hits, 'misses': misses, 'incomplete': incomplete, 'missed_opportunities': missed, 'passed_correctly': passed_ok, 'pending_review': pending, 'deferred_megacap_review': { 'missed': deferred_missed, 'passed_correctly': deferred_ok, 'pending': deferred_pending, }, 'lessons': lessons, 'lessons_for_next_scan': lessons_for_next_scan, 'notes': [ f'Missed-opportunity tracking only covers daily scans from {CANDIDATE_TRACKING_START} onward -- earlier scan history was not retained.', f'A passed-on candidate is flagged "missed" once it is at least {MISSED_OPP_THRESHOLD_PCT:.0f}% above its scan-day price and at least {MIN_DAYS_BEFORE_JUDGING} calendar days have passed.', 'Only new_buys candidates are tracked for missed opportunities -- watchlist/excluded tickers and anything the screeners never surfaced are out of scope, EXCEPT excluded entries tagged funnel_breadth.recurring_megacap_handling (see deferred_megacap_review) -- those carry a price_cad captured at deferral time specifically so a hand-waved name that moved doesn\'t go unnoticed. Coverage starts 2026-08-10, when price_cad capture was added.', 'lessons_for_next_scan is read and applied by daily_scan_prompt.md Step 1a -- this is not just a dashboard display, it changes what the next scan does.', 'unconfirmed_criteria_patterns tallies which entry_setups scoring criterion watchlist/excluded entries most often cite as still-missing, across all scans since ' + CANDIDATE_TRACKING_START + ' -- added 2026-08-12 after volume was found to be the blocker in 20 of 31 such entries, which traced back to Step 3\'s confirmation search never asking for volume at all (now fixed).', ], } with open(OUTPUT_FILE, 'w') as f: json.dump(output, f, indent=2) top_unconfirmed = lessons['unconfirmed_criteria_patterns'][0] if lessons['unconfirmed_criteria_patterns'] else None top_unconfirmed_str = ( f"{top_unconfirmed['criterion']}={top_unconfirmed['count']}/{top_unconfirmed['total_scored_entries']}" if top_unconfirmed else 'none' ) print( f'[generate_retrospective] hits={len(hits)} misses={len(misses)} ' f'incomplete={len(incomplete)} missed_opps={len(missed)} ' f'deferred_megacap_missed={len(deferred_missed)} ' f'realized_pl=${realized_total} lessons={len(lessons_for_next_scan)} ' f'distinct_tickers={lessons["distinct_ticker_count"]} ' f'top_unconfirmed_criterion={top_unconfirmed_str}' ) if __name__ == '__main__': main()