305 lines
17 KiB
Python
305 lines
17 KiB
Python
"""后台准备建议；人工采纳只做限时、无模型的执行检查。"""
|
||
from datetime import datetime, timedelta
|
||
import logging
|
||
import re
|
||
from app.core.plan_revision import compare, fingerprint, portfolio_context, confirmation_revision
|
||
|
||
logger = logging.getLogger('pms.proposal.prepare')
|
||
|
||
|
||
def current_basis(hard, code, proposal_id=None, *, record=True):
|
||
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)
|
||
options = {'force': True} if record else {'record': False, 'plan_override': plan_feed.read_current_plan()}
|
||
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')],
|
||
route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code, **options)
|
||
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': '当前研究资料暂不可用，恢复后自动更新建议', 'stage': 'source'}
|
||
state = selected.get('refresh_status') or {}
|
||
if code in (state.get('pending_codes') or []):
|
||
return {'ok': False, 'error': '这只股票的研究正在更新，完成后重新开放采纳', 'stage': 'research_pending'}
|
||
if (selected.get('consumption') or {}).get('error'):
|
||
return {'ok': False, 'error': '建议保存未完成，系统将自动重试', 'stage': 'record'}
|
||
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': '当前研究资料无法核对，恢复后自动更新建议', 'stage': 'source'}
|
||
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))
|
||
changed = {'research_changed': result == 'material',
|
||
'params_changed': hard.get('scan_params_hash') != ph,
|
||
'portfolio_changed': hard.get('portfolio_context_hash') != context}
|
||
material = any(changed.values())
|
||
return {'ok': not material, 'material': material, **changed,
|
||
'error': '建议依据已变化，后台正在更新' if material else None,
|
||
'view': view, 'stock': stock, 'params': params, 'commitments': commitments,
|
||
'provenance': {'decision': current['decision'], 'consumption_id': current['consumption_id'],
|
||
'version_check': {'status': 'passed', 'at': datetime.now().astimezone().isoformat(),
|
||
'reason': '依据已核对，名次变化不撤销'}}}
|
||
|
||
|
||
def _judgement(p):
|
||
h = p.get('hard_numbers') or {}
|
||
return h.get('prepared_judgement') or {
|
||
'verdict': p.get('judge_verdict') or h.get('judge_verdict') or 'UNAVAILABLE',
|
||
'reason': p.get('judge_reason') or '', 'confidence': h.get('judge_conf'),
|
||
'degraded': (p.get('judge_verdict') or h.get('judge_verdict')) != 'PASS',
|
||
'pv_heads': h.get('judge_pv_heads')}
|
||
|
||
|
||
def ready_hard(hard, verdict, now=None):
|
||
h = dict(hard)
|
||
h.pop('last_accept_check', None)
|
||
h['prepared_judgement'] = {k: verdict.get(k) for k in
|
||
('verdict', 'reason', 'confidence', 'degraded', 'pv_heads')}
|
||
h['preparation'] = {'status': 'ready', 'at': (now or datetime.now()).astimezone().isoformat()}
|
||
return h
|
||
|
||
|
||
def prepare_new(hard, code, verdict):
|
||
"""模型返回期间依据可能变化；只在准备完成后开放确认。"""
|
||
try:
|
||
check = current_basis(hard, code)
|
||
if check.get('ok') and hard.get('consumption_id') and not hard.get('consumption_error'):
|
||
return ready_hard(dict(hard, **check['provenance']), verdict)
|
||
reason = check.get('error') or '建议依据待复核，后台正在更新'
|
||
except Exception:
|
||
logger.exception('新提议准备失败 code=%s', code)
|
||
reason = '建议资料暂不可用，恢复后自动更新'
|
||
return dict(hard, prepared_judgement={k: verdict.get(k) for k in
|
||
('verdict', 'reason', 'confidence', 'degraded', 'pv_heads')},
|
||
preparation={'status': 'pending', 'reason': reason})
|
||
|
||
|
||
def _mark_pending(p, reason, stage='pending'):
|
||
"""先关闭旧内容的确认入口；原JSON条件更新与人工确认互斥。"""
|
||
from app.repo import pms_repo
|
||
h = dict(p.get('hard_numbers') or {})
|
||
state = {'status': 'pending', 'reason': reason, 'stage': stage}
|
||
if h.get('preparation') == state:
|
||
return p
|
||
h['preparation'] = state
|
||
if not pms_repo.record_proposal_check(p, h):
|
||
return None
|
||
return dict(p, hard_numbers=h, hard_numbers_json=pms_repo._dumps(h))
|
||
|
||
|
||
def _pending_response(p, reason, stage='pending'):
|
||
saved = _mark_pending(p, reason, stage)
|
||
return {'ok': False, 'pending': True, 'error': reason,
|
||
'proposal': saved or p, 'proposal_id': p.get('proposal_id')}
|
||
|
||
|
||
def fast_checks(p, check):
|
||
"""核对已展示数量是否仍可执行，不重算建议，不调用模型。"""
|
||
from app.services import proposal_service as ps, param_store, portfolio, market
|
||
from app.core import tradedays as td
|
||
from app.services.execution_conflicts import buy_blocker
|
||
h = p.get('hard_numbers') or {}
|
||
view, params, stock = check['view'], dict(check['params']), check['stock']
|
||
code = p['ts_code']
|
||
conflict = buy_blocker(code)
|
||
if conflict:
|
||
return conflict
|
||
if param_store.get('PMS_AUTONOMY', 'propose_only') == 'off' or param_store.get('PMS_OPEN_AUTONOMY', 'full') == 'off':
|
||
return '自主建仓已暂停'
|
||
t, vp = view['totals'], view['params']
|
||
inflight = check['commitments']
|
||
if int(vp['max_names']) - int(t['names_count']) - len(inflight['names']) <= 0:
|
||
return '可用持仓名额不足，后台将重新安排建议'
|
||
price = float(market.get_price(code) or 0)
|
||
amount = int(p['qty']) * price
|
||
room = float(vp['portfolio_cap']) * float(vp['scale']) - float(t['portfolio_mv']) - inflight['amount']
|
||
if amount > room:
|
||
return '可用仓位额度不足，后台将更新建议数量'
|
||
if t.get('cash_source') == portfolio.CASH_WS and t.get('cash_avail') is not None:
|
||
if amount > float(t['cash_avail']):
|
||
return '可用资金不足，后台将更新建议数量'
|
||
elif param_store.get_bool('PMS_OPEN_REQUIRE_WS_CASH', True):
|
||
return '账户可用资金暂不可读，恢复后重新开放采纳'
|
||
params['_mkt'] = {code: {'day': market.day_snapshot(code) or {}, 'ma5': market.get_ma5(code)}}
|
||
c = {'ts_code': code, 'action': 'OPEN', 'side': 'buy', 'qty': p['qty'],
|
||
'price': price, 'sector': h.get('sector')}
|
||
gate = ps.check_candidate_rules(c, view, params, stock,
|
||
td.ymd() < param_store.get_int('PMS_BRAKE_UNTIL', 0))
|
||
if not gate['passed']:
|
||
return '；'.join(re.sub(r'^[A-Z][A-Z0-9_]*:\s*', '', x) for x in gate['failed'])
|
||
return None
|
||
|
||
|
||
def before_accept(p, expected_revision):
|
||
"""只确认已准备的建议。来源变化交后台，HTTP请求绝不运行完整研判。"""
|
||
hard = p.get('hard_numbers') or {}
|
||
if p.get('action') != 'OPEN' or hard.get('kind') == 'strategy':
|
||
return {'ok': True, 'proposal': p}
|
||
from app.services import proposal_service as ps
|
||
if ps.is_expired(p):
|
||
return {'ok': False, 'error': '建议已过期，不能采纳'}
|
||
if (hard.get('preparation') or {}).get('status') != 'ready' or not hard.get('consumption_id'):
|
||
return {'ok': False, 'pending': True, 'error': (hard.get('preparation') or {}).get('reason')
|
||
or '建议正在准备，完成后可采纳', 'proposal': p}
|
||
if not hard.get('proposal_revision') or expected_revision != hard['proposal_revision']:
|
||
return {'ok': False, 'error': '建议内容已更新，请查看后再次确认', 'proposal': p}
|
||
try:
|
||
check = current_basis(hard, p['ts_code'], proposal_id=p.get('proposal_id'), record=False)
|
||
if check.get('withdraw'):
|
||
return _failed_recheck(p, {'rejected': [{'by': 'qualification', 'why': check['error']}]})
|
||
if not check.get('view'):
|
||
return _pending_response(p, check.get('error') or '建议资料暂不可用', check.get('stage', 'source'))
|
||
if check.get('research_changed') or check.get('params_changed') or hard.get('last_accept_check'):
|
||
return _pending_response(p, '建议依据已变化，后台更新完成后可确认', 'changed')
|
||
reason = fast_checks(p, check)
|
||
if reason:
|
||
return _pending_response(p, reason, 'execution')
|
||
# 保留生成时的已落库消费凭据。采纳核对结果随动作账本写入，不写另一份消费快照。
|
||
checked = check['provenance']
|
||
version_check = dict(checked.get('version_check') or {}, checked_decision=checked.get('decision'))
|
||
return {'ok': True, 'proposal': dict(p, hard_numbers=dict(hard, version_check=version_check))}
|
||
except Exception:
|
||
logger.exception('采纳快速检查失败 proposal=%s', p.get('proposal_id'))
|
||
reason = '当前资料暂不可用，建议保留，后台将自动检查'
|
||
try:
|
||
return _pending_response(p, reason, 'source')
|
||
except Exception:
|
||
logger.exception('保存建议准备状态失败 proposal=%s', p.get('proposal_id'))
|
||
return {'ok': False, 'pending': True, 'error': reason, 'proposal': p}
|
||
|
||
|
||
def refresh_pending(*, now=None, deadline=None):
|
||
"""复用正常自主扫描的交易时段和模型预算；不新增任务。"""
|
||
from app.services import proposal_service as ps
|
||
from app.repo import pms_repo
|
||
results = []
|
||
rows = pms_repo.list_proposals(statuses=('WAIT_USER',), limit=200)
|
||
for p in rows:
|
||
if p.get('action') != 'OPEN' or (p.get('hard_numbers') or {}).get('kind') == 'strategy' or ps.is_expired(p, now):
|
||
continue
|
||
try:
|
||
result = refresh_one(p, now=now, deadline=deadline)
|
||
except Exception:
|
||
logger.exception('后台更新建议失败 proposal=%s', p['proposal_id'])
|
||
result = _pending_response(p, '建议更新暂未完成，系统将自动重试', 'source')
|
||
results.append({'proposal_id': p['proposal_id'], 'ok': result.get('ok', False),
|
||
'withdrawn': result.get('withdrawn', False), 'error': result.get('error')})
|
||
return results
|
||
|
||
|
||
def refresh_one(p, *, now=None, deadline=None):
|
||
from app.services import proposal_service as ps, param_store
|
||
from app.repo import pms_repo
|
||
from app.core import tradedays as td
|
||
now = now or datetime.now()
|
||
hard = p.get('hard_numbers') or {}
|
||
lease = (hard.get('preparation') or {}).get('lease_until')
|
||
if lease and datetime.fromisoformat(lease) > now.astimezone():
|
||
return {'ok': False, 'error': '后台正在更新建议'}
|
||
check = current_basis(hard, p['ts_code'], proposal_id=p['proposal_id'], record=False)
|
||
if check.get('withdraw'):
|
||
return _failed_recheck(p, {'rejected': [{'by': 'qualification', 'why': check['error']}]})
|
||
if not check.get('view'):
|
||
return _pending_response(p, check.get('error') or '建议资料暂不可用', check.get('stage', 'source'))
|
||
if check.get('ok') and (hard.get('preparation') or {}).get('status') == 'ready' and not hard.get('last_accept_check'):
|
||
return {'ok': True}
|
||
research_changed = check.get('research_changed', check.get('material', False))
|
||
previous_failure = hard.get('last_accept_check') or {}
|
||
needs_model = research_changed or previous_failure.get('stage') in ('judge', 'system')
|
||
claimed_hard = dict(hard, preparation={'status': 'pending', 'reason': '建议正在更新，完成后可确认',
|
||
'stage': 'updating', 'lease_until': (now.astimezone() + timedelta(seconds=240)).isoformat()})
|
||
claimed = (dict(p, hard_numbers=claimed_hard, hard_numbers_json=pms_repo._dumps(claimed_hard))
|
||
if pms_repo.record_proposal_check(p, claimed_hard) else None)
|
||
if not claimed:
|
||
return {'ok': False, 'error': '建议已被其他操作更新'}
|
||
p = claimed
|
||
if needs_model and not ps._judge_budget_left(deadline):
|
||
return _pending_response(p, '等待下一轮完成研究更新', 'budget')
|
||
out = {'skipped': [], 'rejected': [], 'queued': [], 'executed': []}
|
||
params = dict(check['params'])
|
||
mkt = {}
|
||
params['_mkt'] = mkt
|
||
fresh = ps._scan_open(check['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'] = needs_model
|
||
verdict = None if needs_model else _judgement(p)
|
||
result = ps._route_one(c, check['view'], params, check['stock'],
|
||
td.ymd() < param_store.get_int('PMS_BRAKE_UNTIL', 0), now, True, out,
|
||
review_only=True, deadline=deadline, prepared_verdict=verdict)
|
||
if not result:
|
||
return _failed_recheck(p, out)
|
||
hn = c['hard_numbers']
|
||
latest = current_basis(hn, p['ts_code'], proposal_id=p['proposal_id'])
|
||
if not latest.get('ok'):
|
||
return _pending_response(p, latest.get('error') or '建议依据再次变化，下一轮继续更新', latest.get('stage', 'changed'))
|
||
hn = ready_hard(dict(hn, **latest['provenance']), result['verdict'], now)
|
||
hn['proposal_revision'] = confirmation_revision(c['qty'], hn, result['verdict'])
|
||
# 仅资金上下文变化而建议内容不变，维持已有确认标识。
|
||
if not needs_model and confirmation_revision(p['qty'], hard, _judgement(p)) == hn['proposal_revision']:
|
||
hn['proposal_revision'] = hard.get('proposal_revision') or hn['proposal_revision']
|
||
c['hard_numbers'] = hn
|
||
if not pms_repo.publish_prepared_proposal(p, c, result['verdict']):
|
||
return {'ok': False, 'error': '建议已被其他操作更新'}
|
||
return {'ok': True, 'proposal': dict(p, qty=c['qty'], hard_numbers=hn,
|
||
judge_verdict=result['verdict'].get('verdict'), judge_reason=result['verdict'].get('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,
|
||
preparation={'status': 'pending', 'reason': message, 'stage': stage})
|
||
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
|