143 lines
8.4 KiB
Python
143 lines
8.4 KiB
Python
"""采纳前核验；复用原扫描、合议、研判与规则闸，不解释自然语言。"""
|
||
from datetime import datetime
|
||
import re
|
||
from app.core.plan_revision import compare, fingerprint, portfolio_context
|
||
|
||
|
||
def current_basis(hard, code, proposal_id=None):
|
||
from app.services import plan_feed, portfolio, command_service, proposal_service as ps
|
||
view = portfolio.positions_view()
|
||
stock = command_service.effective_stock_params()
|
||
params = ps._scan_params(view)
|
||
selected = plan_feed.candidates(held=[x['ts_code'] for x in view['held']],
|
||
black=[c for c, v in stock.items() if v.get('black')], force=True,
|
||
route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code)
|
||
current = next((r for r in selected['items'] if r['ts_code'] == code), None)
|
||
if not (selected.get('decision') or {}).get('id'):
|
||
return {'ok': False, 'error': '当前计划没有可核验版本'}
|
||
state = selected.get('refresh_status') or {}
|
||
if code in (state.get('pending_codes') or []) or (selected.get('consumption') or {}).get('error'):
|
||
return {'ok': False, 'error': '当前计划修订或消费留痕尚未完成'}
|
||
if current is None:
|
||
return {'ok': False, 'error': '该票已不满足当前候选资格', 'material': True, 'withdraw': True}
|
||
result = compare(hard.get('plan_basis'), current, code)
|
||
if result == 'unavailable':
|
||
return {'ok': False, 'error': '当前计划修订或消费留痕尚未完成'}
|
||
ph = fingerprint({k: v for k, v in params.items() if not k.startswith('_')})
|
||
commitments = ps._inflight_open_commitments(view, exclude_proposal=proposal_id)
|
||
context = portfolio_context(view, commitments, stock.get(code))
|
||
material = (result == 'material' or hard.get('scan_params_hash') != ph
|
||
or hard.get('portfolio_context_hash') != context)
|
||
return {'ok': not material, 'material': material,
|
||
'error': '研究、资格或仓位参数已变化，须重新评估' if material else None,
|
||
'view': view, 'stock': stock, 'params': params,
|
||
'provenance': {'decision': current['decision'], 'consumption_id': current['consumption_id'],
|
||
'version_check': {'status': 'passed', 'at': datetime.now().astimezone().isoformat(),
|
||
'reason': '依据相同，名次变化不撤销'}}}
|
||
|
||
|
||
def _failed_recheck(p, out):
|
||
"""明确否决关闭旧提议；暂时性失败保留。原批准依据始终不升级。"""
|
||
from app.repo import pms_repo
|
||
def relevant(rows):
|
||
return [r for r in rows if not r.get('informational') and
|
||
(not r.get('ts_code') or r['ts_code'] == p['ts_code'])]
|
||
rejected = relevant(out.get('rejected', []))
|
||
rows = rejected or relevant(out.get('skipped', []))
|
||
reasons = []
|
||
for row in rows:
|
||
values = row.get('failed') or [row.get('why')]
|
||
if isinstance(values, str):
|
||
values = [values]
|
||
for value in values:
|
||
text = re.sub(r'^[A-Z][A-Z0-9_]*:\s*', '', str(value or '')).strip()
|
||
if text and text not in reasons:
|
||
reasons.append(text)
|
||
stage = rows[0].get('by', 'scan') if rows else 'scan'
|
||
title = {'rule': '交易条件不满足', 'judge': '最新研判不建议建仓',
|
||
'system': '当前策略不建议建仓', 'qualification': '候选条件已变化'}.get(stage, '暂不满足建仓条件')
|
||
message = title + ('：' + '；'.join(reasons) if reasons else '，请稍后查看更新的建议')
|
||
withdrawn = bool(rejected and stage in ('judge', 'system', 'qualification'))
|
||
# 已有检查收尾保留原发生时刻；不假造一次新的研判。
|
||
old_check = (p.get('hard_numbers') or {}).get('last_accept_check') or {}
|
||
at = old_check.get('at') if out is old_check.get('details') else None
|
||
check = {'status': 'withdrawn' if withdrawn else 'blocked',
|
||
'at': at or datetime.now().astimezone().isoformat(),
|
||
'stage': stage, 'message': message, 'reasons': reasons, 'details': out}
|
||
hard = dict(p.get('hard_numbers') or {}, last_accept_check=check)
|
||
save = pms_repo.withdraw_proposal if withdrawn else pms_repo.record_proposal_check
|
||
if not save(p, hard):
|
||
return {'ok': False, 'error': '提议已被更新，请刷新后查看当前结果',
|
||
'proposal_id': p['proposal_id']}
|
||
return {'ok': False, 'withdrawn': withdrawn,
|
||
'error': ('提议已撤下。' if withdrawn else '') + message,
|
||
'details': out, 'proposal_id': p['proposal_id'],
|
||
'proposal': dict(p, hard_numbers=hard, status='EXPIRED' if withdrawn else p.get('status', 'WAIT_USER'))}
|
||
|
||
|
||
def withdraw_recorded_rejections():
|
||
"""旧失败记录由既有扫描收尾。只读结构化否决，不发研判、不按文字猜测。"""
|
||
from app.repo import pms_repo
|
||
count = 0
|
||
for p in pms_repo.list_proposals(statuses=('WAIT_USER',), limit=200):
|
||
hard = p.get('hard_numbers') or {}
|
||
check = hard.get('last_accept_check') or {}
|
||
if (p.get('action') == 'OPEN' and hard.get('kind') != 'strategy' and
|
||
check.get('status') == 'blocked' and check.get('stage') in ('judge', 'system') and
|
||
(check.get('details') or {}).get('rejected')):
|
||
count += bool(_failed_recheck(p, check['details']).get('withdrawn'))
|
||
return count
|
||
|
||
|
||
def before_accept(p, expected_revision):
|
||
from app.services import proposal_service as ps
|
||
from app.repo import pms_repo
|
||
hard = p.get('hard_numbers') or {}
|
||
if p.get('action') != 'OPEN' or hard.get('kind') == 'strategy':
|
||
return {'ok': True, 'proposal': p}
|
||
try:
|
||
check = current_basis(hard, p['ts_code'], proposal_id=p.get('proposal_id'))
|
||
# 已有失败结论不能因仓位后来恰好回到旧值而沿用旧批准。下一次仍完整核验。
|
||
if check.get('ok') and hard.get('last_accept_check'):
|
||
check.update(ok=False, material=True)
|
||
if check.get('ok'):
|
||
if hard.get('proposal_revision') and expected_revision != hard['proposal_revision']:
|
||
return {'ok': False, 'error': '提议内容已更新，请刷新后再次确认', 'proposal': p}
|
||
return {'ok': True, 'proposal': dict(p, hard_numbers=dict(hard, **check['provenance']))}
|
||
if check.get('withdraw'):
|
||
return _failed_recheck(p, {'rejected': [{'by': 'qualification', 'why': check['error']}]})
|
||
if not check.get('material') or not check.get('view'):
|
||
return check
|
||
now = datetime.now()
|
||
out = {'skipped': [], 'rejected': [], 'queued': [], 'executed': []}
|
||
params, view = check['params'], check['view']
|
||
mkt = {}
|
||
params['_mkt'] = mkt
|
||
fresh = ps._scan_open(view, params, check['stock'], {}, mkt, out, now=now,
|
||
review_code=p['ts_code'], exclude_proposal=p['proposal_id'])
|
||
if not fresh:
|
||
return _failed_recheck(p, out)
|
||
c = fresh[0]
|
||
c['_plan_recheck'] = True
|
||
from app.services import param_store
|
||
from app.core import tradedays as td
|
||
brake = td.ymd() < param_store.get_int('PMS_BRAKE_UNTIL', 0)
|
||
result = ps._route_one(c, view, params, check['stock'], brake, now, True, out, review_only=True)
|
||
if not result:
|
||
return _failed_recheck(p, out)
|
||
hn = c['hard_numbers']
|
||
hn['version_check'] = {'status': 'reconfirmed_required', 'at': now.astimezone().isoformat(),
|
||
'reason': '依据变化，已重跑合议、研判、数量及规则闸'}
|
||
hn['proposal_revision'] = fingerprint({'qty': c['qty'], 'hard_numbers': hn})
|
||
if not pms_repo.revise_pending_proposal(p, c, result['verdict']):
|
||
return {'ok': False, 'error': '提议被并发更新，请刷新'}
|
||
pms_repo.insert_ledger(ts_code=p['ts_code'], action=p['action'], arbiter='rule',
|
||
verdict='WARN', price_at=hn.get('price') or 0, hard_numbers=hn,
|
||
ref_id=p['proposal_id'], reason='新版依据变化，更新提议并要求重新确认')
|
||
return {'ok': False, 'error': '提议已按新版重新评估，请查看更新内容后再次确认',
|
||
'proposal': dict(p, qty=c['qty'], hard_numbers=hn,
|
||
judge_verdict=result['verdict'].get('verdict'),
|
||
judge_reason=result['verdict'].get('reason'))}
|
||
except Exception as e:
|
||
return {'ok': False, 'error': f'版本核验暂不可用，保留提议：{type(e).__name__}'}
|