"""目标状态在hold/rebalance下的数量与丢弃记录;无产品运行或收益优化。""" from decimal import Decimal as D import json def run(targets,prices,mode,lot=1,commission=1): if mode not in ('hold','rebalance'):raise ValueError('unknown_mode') if len(targets)!=len(prices):raise ValueError('alignment') if type(lot) is not int or lot<=0:raise ValueError('invalid_lot') fee=D(str(commission)) if not fee.is_finite() or fee<0:raise ValueError('invalid_fee') cash,quantity,fees,last=D(1000),0,D(0),None;rows=[] for i,(raw,raw_price) in enumerate(zip(targets,prices)): price=D(str(raw_price)) if not price.is_finite() or price<=0:raise ValueError('invalid_price') target=D(str(raw)) if raw is not None else None if target is not None and (not target.is_finite() or not 0<=target<=1):raise ValueError('long_only_target') equity=cash+quantity*price;orders=[];dropped=None;reason='no_decision' desired=None if target is not None: desired=int((equity*target//(price*lot)))*lot changed=last is not None and abs(target-last)>D('.000001') if mode=='hold' and quantity and target>0: if changed:dropped={'previous':str(last),'requested':str(target),'reason':'hold_same_direction'} reason='hold_same_direction' if changed else 'unchanged_target' else: delta=desired-quantity reason='rounded_same_size' if delta==0 else 'filled' if delta>0: cost=delta*price+fee if cost>cash:reason='insufficient_cash' else:cash-=cost;quantity+=delta;fees+=fee;orders.append(['buy',delta,str(price),str(fee)]) elif delta<0: cash+=(-delta)*price-fee;quantity+=delta;fees+=fee;orders.append(['sell',-delta,str(price),str(fee)]) last=target rows.append({'i':i,'target':str(target) if target is not None else None,'mode':mode, 'equity_before':str(equity),'desired_rounded_quantity':desired, 'orders':orders,'dropped':dropped,'reason':reason,'quantity':quantity, 'cash':str(cash),'fees':str(fees),'equity_after':str(cash+quantity*price)}) return rows targets=['.4','.8','.2','0'];prices=[10]*4 hold=run(targets,prices,'hold');rebalance=run(targets,prices,'rebalance') assert [r['quantity'] for r in hold]==[40,40,40,0] assert [r['quantity'] for r in rebalance]==[40,79,19,0] assert [r['orders'] for r in hold]==[[['buy',40,'10','1']],[],[],[['sell',40,'10','1']]] assert [r['orders'] for r in rebalance]==[[['buy',40,'10','1']],[['buy',39,'10','1']], [['sell',60,'10','1']],[['sell',19,'10','1']]] assert [r['cash'] for r in hold]==['599','599','599','998'] and hold[-1]['fees']=='2' assert [r['cash'] for r in rebalance]==['599','208','807','996'] and rebalance[-1]['fees']=='4' assert [r['dropped'] for r in hold if r['dropped']]==[ {'previous':'0.4','requested':'0.8','reason':'hold_same_direction'}, {'previous':'0.8','requested':'0.2','reason':'hold_same_direction'}] assert not any(r['dropped'] for r in rebalance) drift=run(['.4','.4','.4','0'],[10,12,11,11],'hold') assert not any(r['dropped'] for r in drift) and drift[-1]['cash']=='1038' and drift[-1]['fees']=='2' rounded=run(['.4','.401'],[10,10],'rebalance') assert rounded[1]['reason']=='rounded_same_size' and rounded[1]['orders']==[] and rounded[1]['dropped'] is None large_lot=run(targets,prices,'rebalance',lot=10) assert [r['quantity'] for r in large_lot]==[40,70,10,0] assert large_lot[1]['orders']==[['buy',30,'10','1']] and large_lot[-1]['cash']=='996' no_fee=run(targets,prices,'rebalance',commission=0) assert [r['quantity'] for r in no_fee]==[40,80,20,0] and no_fee[-1]['cash']=='1000' unknown=run(['.4',None,'0'],[10]*3,'hold') assert unknown[1]['reason']=='no_decision' and unknown[1]['quantity']==40 assert run(targets[:2],prices[:2],'rebalance')==rebalance[:2] assert run(['1'],[10],'rebalance')[0]['reason']=='insufficient_cash' try:run(targets,prices,'resize') except ValueError:pass else:raise AssertionError('未知模式必须拒绝') print(json.dumps({'hold':hold,'rebalance':rebalance,'unchanged_target_price_drift':drift, 'rounding_control':rounded[1],'lot10_quantities':[r['quantity'] for r in large_lot], 'zero_fee_quantities':[r['quantity'] for r in no_fee], 'scope':'synthetic executable targets; no signal generation or real market-rule integration'}))