"""Daily closing-price drawdowns and next-observation breakeven, no third-party deps.""" import csv import datetime as dt import json import math import statistics from pathlib import Path from zoneinfo import ZoneInfo ROOT = Path(__file__).resolve().parent OUT = ROOT / 'outputs' / '20261019' OUT.mkdir(parents=True, exist_ok=True) ASOF = dt.date(2026, 10, 19) NAMES = {'nasdaq': '纳斯达克综合', 'sp500': '标普500'} def read_yahoo(key): x = json.loads((ROOT / 'raw' / f'yahoo_{key}.json').read_text())['chart']['result'][0] result = {} for t, p in zip(x['timestamp'], x['indicators']['quote'][0]['close']): d = dt.datetime.fromtimestamp(t, ZoneInfo('America/New_York')).date() if p is not None and d <= ASOF: assert math.isfinite(p) and p > 0 assert d not in result result[d] = int(round(p * 100)) return result def read_fred(key): result = {} with (ROOT / 'raw' / f'fred_{key}.csv').open() as f: for row in csv.DictReader(f): d = dt.date.fromisoformat(row['observation_date']) v = next(v for k, v in row.items() if k != 'observation_date') if v not in ('', '.') and d <= ASOF: result[d] = int(round(float(v) * 100)) return result def episodes(dates, prices): """Every strictly-below-running-max spell, including one-observation spells.""" peak, active = 0, False result = [] for i in range(1, len(prices)): if prices[i] >= prices[peak]: if active: result.append(make_event(dates, prices, peak, trough, i)) active = False peak = i # last equal high before the next decline elif not active: active, trough = True, i elif prices[i] < prices[trough]: trough = i # first occurrence if the trough ties if active: result.append(make_event(dates, prices, peak, trough, None)) return result def make_event(dates, prices, p, t, r): end = r if r is not None else len(prices) - 1 return { 'peak_i': p, 'trough_i': t, 'recovery_i': r, 'peak_date': dates[p].isoformat(), 'peak_close': prices[p] / 100, 'first_underwater_date': dates[p + 1].isoformat(), 'trough_date': dates[t].isoformat(), 'trough_close': prices[t] / 100, 'recovery_date': dates[r].isoformat() if r is not None else None, 'recovery_close': prices[r] / 100 if r is not None else None, 'max_drawdown': prices[t] / prices[p] - 1, 'decline_calendar_days': (dates[t] - dates[p]).days, 'decline_trading_days': t - p, 'rebound_calendar_days': (dates[r] - dates[t]).days if r is not None else None, 'rebound_trading_days': r - t if r is not None else None, 'total_calendar_days': (dates[r] - dates[p]).days if r is not None else None, 'total_trading_days': r - p if r is not None else None, 'observed_calendar_days': (dates[end] - dates[p]).days, 'observed_trading_days': end - p, 'end_date': dates[end].isoformat(), 'end_drawdown': prices[end] / prices[p] - 1, 'status': '已解套' if r is not None else '未解套', } def next_breakeven(prices): """First strictly later observation with price >= purchase price. O(n).""" result = [None] * len(prices) stack = [] for i, p in enumerate(prices): while stack and p >= prices[stack[-1]]: result[stack.pop()] = i stack.append(i) return result def verify(dates, prices, events, next_ge): assert len(dates) == len(set(dates)) and dates == sorted(dates) # Independent cumulative maximum spell counting. running, underwater, starts = 0, False, 0 for v in prices: running = max(running, v) flag = v < running starts += flag and not underwater underwater = flag assert starts == len(events) for k, e in enumerate(events): p, t, r = e['peak_i'], e['trough_i'], e['recovery_i'] end = r if r is not None else len(prices) assert prices[p] == max(prices[:p + 1]) assert prices[t] == min(prices[p + 1:end]) assert all(x < prices[p] for x in prices[p + 1:end]) if r is not None: assert prices[r] >= prices[p] assert e['total_calendar_days'] == e['decline_calendar_days'] + e['rebound_calendar_days'] if k: assert events[k - 1]['recovery_i'] <= p # Brute-force a deterministic sample plus every event peak and trough. sample = set(range(0, len(prices), 97)) | {len(prices) - 1} sample |= {e[q] for e in events for q in ('peak_i', 'trough_i')} for i in sample: brute = next((j for j in range(i + 1, len(prices)) if prices[j] >= prices[i]), None) assert brute == next_ge[i], (i, brute, next_ge[i]) return {'status': '通过', 'event_invariants_checked': len(events), 'breakeven_bruteforce_samples': len(sample)} def synthetic_tests(): for prices, expected in [([100,100,99,100], [(1,2,3)]), ([100,90,95,80,100],[(0,3,4)]), ([100,90,90],[(0,1,None)]), ([100,101,102],[])]: dates = [dt.date(2000,1,1) + dt.timedelta(days=i) for i in range(len(prices))] ev = episodes(dates,prices) assert [(x['peak_i'],x['trough_i'],x['recovery_i']) for x in ev] == expected verify(dates,prices,ev,next_breakeven(prices)) def csv_write(name, rows): with (OUT / name).open('w', encoding='utf-8-sig', newline='') as f: writer = csv.writer(f) writer.writerows(rows) def main(): synthetic_tests() analysis = {'asof_date': ASOF.isoformat(), 'indices': {}, 'source_priority': 'FRED有效收盘优先；FRED无观测时使用Yahoo同指数日收盘；保留逐行来源。不填充休市日。', 'checks': {}} diffs = [['指数', '日期', 'Yahoo收盘', 'FRED收盘', '差值Yahoo减FRED', '最终使用']] for key in NAMES: yahoo, fred = read_yahoo(key), read_fred(key) # FRED has an anomalous quote on Good Friday 2019-04-19. # Use the matching Yahoo index's observed-session dates as the date spine. # This preserves real exchange closures instead of inventing a session. excluded = [{'date':d.isoformat(),'fred_close':fred[d]/100,'reason':'FRED有值但Yahoo无交易观测；2019-04-19为Good Friday休市'} for d in sorted(fred.keys()-yahoo.keys())] assert all(x['date'] == '2019-04-19' for x in excluded), excluded canonical = {d:fred.get(d,p) for d,p in yahoo.items()} dates = sorted(canonical) prices = [canonical[d] for d in dates] sources = ['FRED' if d in fred else 'Yahoo Finance' for d in dates] ev, nge = episodes(dates, prices), next_breakeven(prices) mismatches = [(d, yahoo[d], fred[d]) for d in sorted(yahoo.keys() & fred.keys()) if yahoo[d] != fred[d]] diffs += [[NAMES[key], d.isoformat(), a/100, b/100, (a-b)/100, 'FRED'] for d,a,b in mismatches] check = verify(dates, prices, ev, nge) complete = [e for e in ev if e['recovery_i'] is not None] buckets = [] for cutoff in [0, .01, .05, .10, .20, .30, .50]: subset = [e for e in ev if -e['max_drawdown'] >= cutoff] done = [e for e in subset if e['recovery_i'] is not None] times = [e['total_calendar_days'] for e in done] buckets.append({'cutoff':cutoff,'all_events':len(subset),'recovered':len(done),'open':len(subset)-len(done),'median_calendar_days':statistics.median(times) if times else None,'mean_calendar_days':statistics.mean(times) if times else None,'max_calendar_days':max(times) if times else None}) daily = [] ath = 0 for i,(d,p,s) in enumerate(zip(dates,prices,sources)): ath = max(ath,p) j = nge[i] last = j if j is not None else len(dates)-1 # A price below purchase before first >= means an actual loss spell. loss = i + 1 < len(prices) and prices[i+1] < p status = '最后观测日' if i == len(prices)-1 else ('已回本' if j is not None else '尚未回本') daily.append([NAMES[key],d.isoformat(),p/100,s,ath/100,p/ath-1,dates[j].isoformat() if j is not None else None,(dates[j]-d).days if j is not None else None,j-i if j is not None else None,'是' if loss else '否',status,(dates[last]-d).days,last-i,i]) longest = max(complete,key=lambda x:x['total_calendar_days']) deepest = min(ev,key=lambda x:x['max_drawdown']) item = {'name':NAMES[key],'key':key,'start':dates[0].isoformat(),'end':dates[-1].isoformat(),'observations':len(dates),'fred_rows':sources.count('FRED'),'yahoo_rows':sources.count('Yahoo Finance'),'source_overlap':len(yahoo.keys()&fred.keys()),'source_mismatches':len(mismatches),'excluded_observations':excluded,'events':ev,'buckets':buckets,'daily':daily,'longest':longest,'deepest':deepest,'open_events':[e for e in ev if e['recovery_i'] is None]} analysis['indices'][key] = item analysis['checks'][key] = check csv_write(f'{key}_daily.csv', [['指数','日期','收盘点位','来源','历史最高收盘','距历史最高回撤','买入后首次不低于买价日期','等待自然日','等待交易日','等待期间发生浮亏','状态','截至回本或样本末已观察自然日','已观察交易日','交易日序号从0起']] + daily) fields = ['peak_date','first_underwater_date','peak_close','trough_date','trough_close','recovery_date','recovery_close','max_drawdown','decline_calendar_days','decline_trading_days','rebound_calendar_days','rebound_trading_days','total_calendar_days','total_trading_days','status','end_date','observed_calendar_days','observed_trading_days','end_drawdown'] header = ['前高日','首次跌破日','前高点位','谷底日','谷底点位','解套日','解套收盘','最大回撤','下跌自然日','下跌交易日','谷底后恢复自然日','谷底后恢复交易日','总解套自然日','总解套交易日','状态','观测截止日','已观察自然日','已观察交易日','截止日距前高回撤'] csv_write(f'{key}_drawdowns.csv', [header] + [[e[f] for f in fields] for e in ev]) print(json.dumps({k:v for k,v in item.items() if k not in ('daily','events')},ensure_ascii=False,indent=2)) csv_write('source_discrepancies.csv',diffs) (ROOT/'analysis.json').write_text(json.dumps(analysis,ensure_ascii=False,separators=(',',':'))) (OUT/'validation.json').write_text(json.dumps(analysis['checks'],ensure_ascii=False,indent=2)) if __name__ == '__main__': main()