"""冻结披露版本、四个非重叠季度、资本开支符号与20股季度现金账。""" from copy import deepcopy from decimal import Decimal as D import json def make_facts(): facts = [] for i in range(1, 23): for end, available, quarters in [ ('2025Q4', '2026-02-15', ['2025Q1','2025Q2','2025Q3','2025Q4']), ('2026Q1', '2026-05-15', ['2025Q2','2025Q3','2025Q4','2026Q1'])]: cfo = [10,10,10,10+i] if end == '2025Q4' else [10,10,10+i,10] if i == 21: cfo = [2,2,3,3] if end == '2025Q4' else [2,3,3,92] facts.append({'symbol':f'S{i:02}', 'period':end, 'available':available, 'quarters':quarters, 'cfo':cfo, 'capex':[-5]*4, 'capex_convention':'negative_outflow', 'financial':i == 22}) return facts def score(fact, cap): ids = [int(q[:4])*4+int(q[-1])-1 for q in fact['quarters']] if len(ids) != 4 or any(b-a != 1 for a,b in zip(ids, ids[1:])): raise ValueError('TTM季度须连续且不重叠') if len(fact['cfo']) != 4 or len(fact['capex']) != 4: raise ValueError('TTM字段数量不符') convention = fact['capex_convention'] if convention not in ('negative_outflow','positive_outflow'): raise ValueError('资本开支符号未定义') multiplier = -1 if convention == 'negative_outflow' else 1 outflows = [D(str(x))*multiplier for x in fact['capex']] values = [D(str(x)) for x in fact['cfo']] cap = D(str(cap)) if not cap.is_finite() or cap <= 0 or any(not x.is_finite() for x in values+outflows): raise ValueError('无效财务值或市值') if any(x < 0 for x in outflows): raise ValueError('符号映射后资本开支为负') fcf = sum(values)-sum(outflows) return fcf/cap if fcf > 0 else None def select(facts, cutoff): selected, lineage = {}, {} for symbol in sorted({f['symbol'] for f in facts}): known = [f for f in facts if f['symbol'] == symbol and f['available'] <= cutoff] if not known: continue fact = max(known,key=lambda f:(f['period'],f['available'])) if fact['financial']: continue value = score(fact,1000) if value is not None: selected[symbol], lineage[symbol] = value,fact['available'] names = sorted(selected,key=lambda s:(-selected[s],s))[:20] return names, selected, lineage facts = make_facts() execution_days=['2026-04-01','2026-07-01','2026-07-02'] assert execution_days[0]>'2026-03-31' and execution_days[1]>'2026-06-30' first, scores, lineage = select(facts,'2026-03-31') second, _, _ = select(facts,'2026-06-30') assert first == [f'S{i:02}' for i in range(20,0,-1)] assert second == ['S21']+[f'S{i:02}' for i in range(20,1,-1)] assert scores['S20'] == D('.04') and 'S21' not in scores assert all(day <= '2026-03-31' for day in lineage.values()) assert select(list(reversed(facts)),'2026-03-31') == select(facts,'2026-03-31') positive = dict(facts[0],capex=[5]*4,capex_convention='positive_outflow') assert score(positive,1000) == score(facts[0],1000) for bad in [dict(facts[0],capex=[5]*4),dict(facts[0],quarters=['2025Q1']*4)]: try: score(bad,1000) except ValueError: pass else: raise AssertionError('符号或季度不符必须拒绝') cash, holdings, fees, orders = D(10000),set(),D(0),[] for stage,target in enumerate([first,second,[]]): for symbol in sorted(holdings-set(target)): price = D(11) if stage == 1 else D('10.2') cash += 40*price-1; fees += 1 orders.append([stage,symbol,'sell',40,str(price),1]) for symbol in sorted(set(target)-holdings): cash -= D(401); fees += 1 orders.append([stage,symbol,'buy',40,'10',1]) holdings = set(target) if stage == 0: assert cash == 1980 and cash+len(holdings)*400 == 10000-fees if stage == 1: assert cash == 2018 and len(holdings) == 20 and fees == 22 assert cash == D('10158.0') and fees == 42 and not holdings assert orders[:20] == [[0,f'S{i:02}','buy',40,'10',1] for i in range(1,21)] assert orders[20:22] == [[1,'S01','sell',40,'11',1],[1,'S21','buy',40,'10',1]] assert orders[22:] == [[2,s,'sell',40,'10.2',1] for s in sorted(second)] print(json.dumps({'first':first,'second':second,'first_yield_S20':str(scores['S20']), 'execution_days_by_stage':execution_days, 'first_lineage':lineage,'orders':orders,'ending_cash':str(cash), 'fees':str(fees),'holdings':sorted(holdings), 'scope':'synthetic frozen financial snapshots, not SEC history download'}))