"""DJ111—DJ120：构造字段、版本、证据校验实验，不调用任何AI模型。""" from datetime import date from decimal import Decimal, ROUND_HALF_UP from hashlib import sha256 from pathlib import Path import json import re import sys def canonical(data): return json.dumps(data,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode('utf-8') def unit_value(raw, unit): factors={'元':1,'万元':10000,'亿元':100000000} if unit not in factors: return None return float(Decimal(str(raw))*factors[unit]) def header_unit(header): # 复合量纲优先，避免将元／股误拆成元。 for unit in ['元／股','亿元','万元','元']: if unit in header:return unit return None def same_context(*cells): keys=['entity','scope','start','end','currency','version'] return all(all(c[k]==cells[0][k] for k in keys) for c in cells[1:]) def parse_cell(raw,dash_zero=False): clean=re.sub(r'[①②③④⑤⑥⑦⑧⑨⑩]$','',raw.strip()) if not clean:return {'state':'missing','value':None} if clean in ['—','-','–']: return {'state':'explicit_zero' if dash_zero else 'needs_legend','value':0 if dash_zero else None} neg=clean.startswith('(') and clean.endswith(')') clean=clean[1:-1] if neg else clean if not re.fullmatch(r'[+-]?\d+(?:,\d{3})*(?:\.\d+)?',clean): return {'state':'reject','value':None} value=Decimal(clean.replace(',',''))*(-1 if neg else 1) return {'state':'numeric','value':float(value)} def calculate(): out={} samples=[('收入表，单位人民币万元','120000','万元','亿元'),('现金表，单位人民币亿元','2.4','亿元','万元'),('费用表，单位人民币元','30000','元','元'),('每股收益，单位元／股','1.2','元／股','元'),('无单位表头','100',None,'万元')] rows=[] for header,raw,expected_unit,ai_unit in samples: unit=header_unit(header) assert unit==expected_unit correct=unit_value(raw,unit); wrong=unit_value(raw,ai_unit) rows.append(dict(header=header,raw=raw,source_unit=unit,standard_unit='元' if correct is not None else None, multiplier={'元':1,'万元':10000,'亿元':100000000}.get(unit),source_standard_value=correct, candidate_unit=ai_unit,candidate_value=wrong,ratio=wrong/correct if correct else None, allowed=correct is not None and unit==ai_unit)) out['DJ111']={'cases':rows} group={'revenue':1000,'cost':700,'net':120,'cfo':150,'assets':2000,'liabilities':800,'equity':1200} parent={'revenue':600,'cost':300,'net':90,'cfo':40,'assets':1200,'liabilities':400,'equity':800} context=dict(entity='构造甲公司',start='2025-01-01',end='2025-12-31',currency='CNY',version='v1') a=dict(context,scope='合并',value=group['cfo']);b=dict(context,scope='母公司',value=parent['net']) out['DJ112']={'group':group,'parent':parent,'group_margin':(group['revenue']-group['cost'])/group['revenue'], 'parent_margin':(parent['revenue']-parent['cost'])/parent['revenue'], 'mixed_margin':(group['revenue']-parent['cost'])/group['revenue'], 'group_cfo_net':group['cfo']/group['net'],'mixed_cfo_net':parent['cfo']/group['net'], 'mixed_context_allowed':same_context(a,b)} assert all(v['assets']==v['liabilities']+v['equity'] for v in [group,parent]) original={'id':'教学2026A001','published':'2026-03-01','period':'2025','scope':'合并','fields':dict(revenue=1000,cost=700,net=120,cfo=150)} correction={'id':'教学2026A002','published':'2026-03-20','corrects':'教学2026A001','period':'2025','scope':'合并','replacements':dict(revenue=900,cost=680,net=80)} snapshots=[] for asof in ['2026-03-10','2026-04-01']: effective=dict(original['fields']);provenance={k:original['id'] for k in effective} if correction['published']<=asof: for k,v in correction['replacements'].items():effective[k]=v;provenance[k]=correction['id'] snapshots.append(dict(asof=asof,fields=effective,provenance=provenance,margin=(effective['revenue']-effective['cost'])/effective['revenue'],cfo_net=effective['cfo']/effective['net'])) out['DJ113']={'original':original,'correction':correction,'original_sha256':sha256(canonical(original)).hexdigest(),'correction_sha256':sha256(canonical(correction)).hexdigest(),'snapshots':snapshots,'late_retrieval_old_file':'2026-04-02'} def days(start,end):return (date.fromisoformat(end)-date.fromisoformat(start)).days+1 a=('2025-01-01','2025-12-31');b=('2025-04-01','2026-03-31') aq=[250,300,300,350];bq=[300,300,350,400] out['DJ114']={'a_period':a,'b_period':b,'a_days':days(*a),'b_days':days(*b),'overlap_days':days(max(a[0],b[0]),min(a[1],b[1])), 'a_quarters':aq,'b_quarters':bq,'a_fiscal_revenue':sum(aq),'b_fiscal_revenue':sum(bq), 'b_calendar_2025_revenue':200+sum(bq[:3]),'b_prior_janmar':200,'wrong_comparison_difference':sum(bq)-sum(aq), 'aligned_difference':200+sum(bq[:3])-sum(aq)} cases=[('1,200',False),('(1,200)',False),('—',False),('—',True),('',False),('(120)①',False),('120①',False),('12O',False)] out['DJ115']={'cases':[dict(raw=raw,dash_zero=legend,**parse_cell(raw,legend)) for raw,legend in cases], 'correct_net':500+parse_cell('(120)')['value']-30, 'sign_dropped_net':500+120-30} cells=[dict(context,scope='合并',page_printed=88,page_pdf=90,row='经营活动现金流量净额',column='2025年度',raw='150.45',unit='万元',value=150.45), dict(context,scope='合并',page_printed=86,page_pdf=88,row='净利润',column='2025年度',raw='120.12',unit='万元',value=120.12)] assert same_context(*cells) ratio=Decimal(cells[0]['raw'])/Decimal(cells[1]['raw']) out['DJ116']={'cells':cells,'formula':'经营活动现金流量净额 / 净利润','ratio':float(ratio),'percentage':float(ratio*100), 'percentage_display':str((ratio*100).quantize(Decimal('.01'),rounding=ROUND_HALF_UP)), 'rounded_inputs_ratio':round(cells[0]['value'])/round(cells[1]['value']), 'context_checks':dict(valid=same_context(*cells),wrong_scope=same_context(cells[0],dict(cells[1],scope='母公司')),wrong_period=same_context(cells[0],dict(cells[1],end='2024-12-31')),wrong_version=same_context(cells[0],dict(cells[1],version='v0')))} docs=[dict(id='教学预告001',published='2026-01-20',state='预告区间',audit='未审计',low=80,high=120,reported=None), dict(id='教学快报001',published='2026-02-20',state='快报值',audit='未审计',reported=95), dict(id='教学年报001',published='2026-03-20',state='定报值',audit='本构造材料明确含审计报告',reported=88)] for item in docs: item.update(entity='构造甲公司',period='2025',scope='合并',metric='归母净利润',unit='万元') item['eligible_for_reported_actual']=item['reported'] is not None out['DJ117']={'documents':docs,'forecast_midpoint':(80+120)/2,'midpoint_is_reported_actual':False, 'flash_to_annual_difference':docs[-1]['reported']-docs[-2]['reported'], 'forecast_range_contains_annual':docs[0]['low']<=docs[-1]['reported']<=docs[0]['high'], 'asof_selection':[dict(asof=day,selected=max((x for x in docs if x['published']<=day),key=lambda x:x['published'])['id']) for day in ['2026-01-25','2026-02-25','2026-03-25']]} required=['date','currency','earnings_basis'] base=dict(price=20,eps=1,date='2026-10-23',currency='CNY',earnings_basis='TTM归母基本每股收益',scope='合并') cards=[] for field in required: item=dict(base);del item[field] missing=[x for x in required if x not in item] cards.append(dict(missing=missing,supplied=item,pe=None if missing else item['price']/item['eps'])) out['DJ118']={'cards':cards,'complete_card':base,'complete_pe':base['price']/base['eps'], 'assumed_price_examples':[dict(price=p,eps=1,pe=p/1) for p in [10,20]], 'assumed_basis_examples':[dict(basis=name,eps=eps,pe=20/eps) for name,eps in [('上年',1),('TTM',.8),('预测',1.25)]], 'assumed_fx_example':dict(price_CNY=20,eps_USD=1,CNY_per_USD=7,pe=20/(1*7))} models=[] for state,capex in [('未披露',None),('明确为零',0),('另列构造估计',40)]: if capex is None:fcf=ev=equity=per_share=None else: fcf=100-capex;ev=fcf*1.02/(.1-.02);equity=ev-200;per_share=equity/100 models.append(dict(state=state,operating_cash=100,capex=capex,fcf=fcf,enterprise_value=ev,equity_value=equity,per_share=per_share)) out['DJ119']={'discount_assumption':.1,'growth_assumption':.02,'net_debt':200,'shares':100,'models':models} original_bytes='构造公告：教学2026A001\n页码：1\n收入：900万元\n成本：680万元\n'.encode('utf-8') modified_bytes=original_bytes.replace(b'900',b'901') newline_bytes=original_bytes.replace(b'\n',b'\r\n') expected=sha256(original_bytes).hexdigest() fixtures=Path(__file__).parent/'fixtures' fixtures.mkdir(exist_ok=True) files=[] for name,data in [('ai-source-v1.txt',original_bytes),('ai-source-changed.txt',modified_bytes),('ai-source-crlf.txt',newline_bytes)]: path=fixtures/name;path.write_bytes(data) digest=sha256(path.read_bytes()).hexdigest() files.append(dict(file=name,byte_length=len(data),sha256=digest,matches_original=(digest==expected))) assert files[0]['matches_original'] and not files[1]['matches_original'] and not files[2]['matches_original'] out['DJ120']={'archive':dict(announcement_id='教学2026A001',public_source=None,source_status='纯构造教学文件，无真实公开公告',download_day=None,generated_day='2026-10-23',page=1,local_file='fixtures/ai-source-v1.txt',sha256=expected),'files':files,'byte_content_utf8':original_bytes.decode('utf-8')} assert len(out)==10 # 对可识别的经济量、语义状态和不变量交叉检查，不能只验证JSON成功写出。 assert rows[0]['source_standard_value'] / rows[0]['candidate_value'] == 1/10000 assert out['DJ112']['mixed_context_allowed'] is False assert snapshots[0]['fields']['cfo']==snapshots[1]['fields']['cfo'] assert snapshots[0]['provenance']['net'] != snapshots[1]['provenance']['net'] assert out['DJ114']['a_days']==out['DJ114']['b_days'] and out['DJ114']['overlap_days']