181 lines
10 KiB
Python
181 lines
10 KiB
Python
"""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()
|