"""R637:同币种多头袖套预算合成,统一上限与现金调整显式。""" from math import isfinite,isclose MAP={'AAA':'A','A_ALIAS':'A','BBB':'B','CCC':'C'} BUDGET={'S1':.6,'S2':.4} CURRENCY={'S1':'USD','S2':'USD'} def vector(values): if any(not isfinite(x) or x<0 for x in values.values()) or abs(sum(values.values())-1)>1e-12: raise ValueError('invalid normalized allocation') def merge(sleeves,budgets,currencies): vector(budgets) if set(sleeves)!=set(budgets):raise ValueError('missing sleeve') if set(currencies)!=set(budgets) or any(not isinstance(c,str) or not c for c in currencies.values()) or len(set(currencies.values()))!=1:raise ValueError('currency mismatch or unknown') contributions={};raw={} for sleeve,weights in sleeves.items(): vector(weights);contributions[sleeve]={} for alias,w in weights.items(): if alias not in MAP:raise ValueError('unknown identity') s=MAP[alias];amount=budgets[sleeve]*w raw[s]=raw.get(s,0)+amount;contributions[sleeve][s]=contributions[sleeve].get(s,0)+amount capped={s:min(w,.5) for s,w in raw.items()};cash=1-sum(capped.values()) assert all(isclose(raw[s],sum(row.get(s,0) for row in contributions.values()),rel_tol=0,abs_tol=1e-12) for s in raw) assert all(isfinite(w) and 0<=w<=.5 for w in capped.values()) assert cash>=-1e-12 and isclose(sum(capped.values())+cash,1,rel_tol=0,abs_tol=1e-12) return raw,capped,cash,contributions policies=[{'S1':{'AAA':.7,'BBB':.3},'S2':{'A_ALIAS':.5,'CCC':.5}}, {'S1':{'AAA':.4,'BBB':.6},'S2':{'A_ALIAS':.25,'CCC':.75}}] prices=[{'A':100.,'B':100.,'C':100.},{'A':110.,'B':100.,'C':90.},{'A':100.,'B':105.,'C':95.}] cash=1000.;shares={s:0. for s in 'ABC'};ledger=[] for t,policy in enumerate(policies): raw,target,cash_weight,contrib=merge(policy,BUDGET,CURRENCY) nav=cash+sum(shares[s]*prices[t][s] for s in shares) desired={s:nav*target.get(s,0)/prices[t][s] for s in shares} fills={s:desired[s]-shares[s] for s in shares} for side in ['sell','buy']: for s,qty in fills.items(): if (side=='sell' and qty>=0) or (side=='buy' and qty<=0):continue cash-=qty*prices[t][s] assert cash>=-1e-8 shares=desired assert cash>=-1e-8 and isclose(cash+sum(shares[s]*prices[t][s] for s in shares),nav,rel_tol=0,abs_tol=1e-8) ledger.append({'raw':raw,'target':target,'cash_weight':cash_weight,'contributions':contrib,'fills':fills,'cash':cash}) ending=cash+sum(shares[s]*prices[-1][s] for s in shares) assert isclose(ledger[0]['raw']['A'],.62,rel_tol=0,abs_tol=1e-12) assert isclose(ledger[0]['cash_weight'],.12,rel_tol=0,abs_tol=1e-12) assert isclose(ending,1033.870303030303,rel_tol=0,abs_tol=1e-8) for bad in [{'A':float('nan'),'B':1},{'A':float('inf'),'B':0}]: try:vector(bad) except ValueError:pass else:raise AssertionError('nonfinite weights accepted') try:merge(policies[0],{'S1':.6,'S2':.6},CURRENCY) except ValueError:pass else:raise AssertionError('sleeve budget above one accepted') try:merge({'S1':{'UNKNOWN':1},'S2':{'CCC':1}},BUDGET,CURRENCY) except ValueError:pass else:raise AssertionError('unknown security silently dropped') try:merge(policies[0],BUDGET,{'S1':'USD','S2':'HKD'}) except ValueError:pass else:raise AssertionError('cross-currency sleeves merged without FX contract') naive={} for sleeve in policies[0].values(): for alias,weight in sleeve.items():naive[MAP[alias]]=naive.get(MAP[alias],0)+weight assert isclose(sum(naive.values()),2,rel_tol=0,abs_tol=1e-12) try:vector(naive) except ValueError:pass else:raise AssertionError('two fully funded strategies added as one budget') for t,row in enumerate(ledger):print('period='+str(t)+' raw='+str(row['raw'])+' capped='+str(row['target'])+' cash_weight='+format(row['cash_weight'],'.6f')+' fills='+str({k:round(v,6) for k,v in row['fills'].items()})) print('ending='+format(ending,'.6f')+' fees=0_synthetic duplicate_A=merged unknown_identity=blocked NaN_and_overbudget=rejected')