diff --git a/DEVLOG.md b/DEVLOG.md index ae932ff..05ac67b 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -2055,3 +2055,19 @@ QMT在线,无补同步要求,健康正常,正常回放0差异。93行指 本地Chrome替身页面零错误,线上服务读回进展正确;生产登录浏览器未直接验收。 人工T+1、失败撤单与并发边界仅离线验证,不制造交易,188未操作。 四仓规格与交接逐字同步,完整备份和回退见14:30交接。 + + +## 2026-10-20 提议先准备后采纳:本地实施 + +用户批准将完整重评移到后台。待确认建议先完成数量、规则和消费记录。 +采纳只读当前完整计划并检查已展示数量,不调用研判、不新增消费快照。 +原自主扫描维护旧建议,研究变化重评,组合与参数变化复用原研判。 +明确否决撤下,来源或留痕失败保持准备中。并发更新采用条件写入。 +准备结果与动作审计同一事务,人工确认不能越过旧内容标识。 +改动涉及plan_revision、plan_feed、plan_recheck、proposal_service、仓储和网页。 +新增batch42共24例,更新batch38共30例。46批本地隔离禁网回归通过。 +三页面守卫通过。本地Chrome替身测试页面错误零,确认提交一次。 +窄屏展开右栏后遮挡确认框的旧问题同步修复,样式保持。 +当前155尚未部署本改动。11:30备份为prepared-before-20261020.json。 +现场没有WAIT_USER提议,不将替身验证当作自然用户采纳判收。 +正式规格见docs/开发规格_提议先准备后采纳_2026-10-20.md。 diff --git a/app/core/plan_revision.py b/app/core/plan_revision.py index 563b0b1..9e58e96 100644 --- a/app/core/plan_revision.py +++ b/app/core/plan_revision.py @@ -45,7 +45,14 @@ def compare(old, current, code): def needs_new_judgement(hard, previous): - """决策系统缓存按股票/动作,缺同依据的既有研判就必须要求重新回答。""" - keys = ('plan_basis', 'scan_params_hash', 'portfolio_context_hash') - return (not previous or any(previous.get(k) is None for k in keys) - or any(fingerprint(hard.get(k)) != fingerprint(previous.get(k)) for k in keys)) + """仅研究依据变化重问模型;组合及参数变化复算规则和数量。""" + return (not previous or previous.get('plan_basis') is None + or fingerprint(hard.get('plan_basis')) != fingerprint(previous.get('plan_basis'))) + + +def confirmation_revision(qty, hard, verdict): + """确认针对可见的建议内容;后台检查时间、名次及无关版本不让确认失效。""" + return fingerprint({'qty': qty, 'basis': hard.get('plan_basis'), + 'advice': hard.get('advice'), 'reason': hard.get('reason'), + 'verdict': verdict.get('verdict'), 'judge_reason': verdict.get('reason'), + 'confidence': verdict.get('confidence')}) diff --git a/app/db/session.py b/app/db/session.py index 1eef764..95a1aff 100644 --- a/app/db/session.py +++ b/app/db/session.py @@ -332,3 +332,36 @@ def withdraw_rejected_proposal(old, hard): 'failed':encode(check.get('reasons') or []), 'reason':('提议已撤下:'+check['message'])[:500], 'ref':old['proposal_id']}) return 1 + + +def publish_prepared_proposal(old, candidate, verdict): + """准备结果与变更审计同成同败。旧确认与后台发布只有一个能赢。""" + import json + from datetime import datetime + encode = lambda v: json.dumps(v, ensure_ascii=False, default=str) + hard = candidate['hard_numbers'] + if (hard.get('preparation') or {}).get('status') != 'ready' or not hard.get('consumption_id'): + raise ValueError('准备结果缺少完整依据或消费凭据') + old_hard = old.get('hard_numbers') or {} + audit = dict(hard, previous_proposal={'qty': old['qty'], 'decision': old_hard.get('decision'), + 'consumption_id': old_hard.get('consumption_id'), 'proposal_revision': old_hard.get('proposal_revision'), + 'judge_verdict': old.get('judge_verdict'), 'judge_reason': old.get('judge_reason')}) + _check_cooldown('proxy') + with get_engine('proxy').begin() as c: + def query(sql, params): + sql = map_tables(sql) + assert_single_table(sql) + return c.execute(text(sql), params) + n = query("UPDATE pms_proposal SET qty=:qty,hard_numbers_json=:hn,judge_verdict=:jv,judge_reason=:jr " + "WHERE proposal_id=:pid AND status='WAIT_USER' AND hard_numbers_json=:old", + {'qty':candidate['qty'],'hn':encode(hard),'jv':verdict.get('verdict'), + 'jr':(verdict.get('reason') or '')[:500],'pid':old['proposal_id'],'old':old['hard_numbers_json']}).rowcount + if not n: + return 0 + query('INSERT INTO pms_action_ledger (ts_code,decided_at,action,arbiter,verdict,price_at,' + 'hard_numbers_json,failed_checks_json,reason,ref_id,outcome_scored) ' + 'VALUES (:code,:at,:action,:arbiter,:verdict,:price,:hard,:failed,:reason,:ref,0)', + {'code':old['ts_code'],'at':datetime.now(),'action':old['action'],'arbiter':'rule', + 'verdict':'NOTE','price':float(hard.get('price') or 0),'hard':encode(audit),'failed':'[]', + 'reason':'后台完成建议更新,等待人工确认','ref':old['proposal_id']}) + return 1 diff --git a/app/repo/pms_repo.py b/app/repo/pms_repo.py index ebf07be..4f0e6ac 100644 --- a/app/repo/pms_repo.py +++ b/app/repo/pms_repo.py @@ -1091,3 +1091,8 @@ def list_op_log(*, limit: int = 200) -> list: for r in rows: r["params"] = _loads(r.get("params_json"), {}) return rows + + +def publish_prepared_proposal(old, candidate, verdict): + from app.db.session import publish_prepared_proposal as publish + return publish(old, candidate, verdict) diff --git a/app/services/plan_feed.py b/app/services/plan_feed.py index fc3e947..a666868 100644 --- a/app/services/plan_feed.py +++ b/app/services/plan_feed.py @@ -901,6 +901,15 @@ def get_plan(*, force: bool = False, date=None) -> dict: return plan +def read_current_plan(): + """采纳只读最新完整发布版,限时取数,不写接收快照或行业映射。""" + p = _params() + plan = fetch(base=p['base'], path=p['path'], timeout=min(3, max(1, p['timeout'])), + extra_params=build_query(p)) + assert_fresh(plan, max_stale_tdays=p['stale_tdays']) + return plan + + def invalidate(): with _lock: _cache.update({"at": 0.0, "key": None, "plan": None, "error": None}) @@ -1147,7 +1156,7 @@ def snapshot_log(*, plan_date=None, limit: int = 50) -> dict: # ================================================================ 对外: 候选与状态 def candidates(*, held=(), black=(), route_by_verdict: bool = False, force=False, - record=True, recheck_code=None) -> dict: + record=True, recheck_code=None, plan_override=None) -> dict: """参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。 route_by_verdict 默认 False: 命令驱动建仓 (command_service._candidates) 不传, 于是用户 @@ -1157,7 +1166,7 @@ def candidates(*, held=(), black=(), route_by_verdict: bool = False, force=False 显式传 True。要让命令驱动也按判决过滤, 改这里的调用方, 不改默认值。 """ p = _params() - plan = get_plan(force=force) if force else get_plan() + plan = plan_override if plan_override is not None else (get_plan(force=force) if force else get_plan()) route = bool(route_by_verdict) out = select_candidates(plan, held=held, black=black, top_n=0 if recheck_code else p["top_n"], tiers=p["tiers"], include_observe=p["include_observe"], @@ -1200,6 +1209,7 @@ def candidates(*, held=(), black=(), route_by_verdict: bool = False, force=False capped_main=False, capped_obs=False) except Exception as e: # 记录失败可展示,不能把无凭据候选变成自主指令。 + logger.exception('候选消费记录保存失败 code=%s decision=%s', recheck_code, (plan.get('decision') or {}).get('id')) out['consumption']['error'] = type(e).__name__ for row in out['items']: row['consumption_error'] = type(e).__name__ diff --git a/app/services/plan_recheck.py b/app/services/plan_recheck.py index 7316987..90e6049 100644 --- a/app/services/plan_recheck.py +++ b/app/services/plan_recheck.py @@ -1,39 +1,253 @@ -"""采纳前核验;复用原扫描、合议、研判与规则闸,不解释自然语言。""" -from datetime import datetime +"""后台准备建议;人工采纳只做限时、无模型的执行检查。""" +from datetime import datetime, timedelta +import logging import re -from app.core.plan_revision import compare, fingerprint, portfolio_context +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): +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')], force=True, - route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code) + 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': '当前计划没有可核验版本'} + return {'ok': False, 'error': '当前研究资料暂不可用,恢复后自动更新建议', 'stage': 'source'} 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 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': '当前计划修订或消费留痕尚未完成'} + 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)) - 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, + 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': '依据相同,名次变化不撤销'}}} + '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') + # 保留生成时的已落库消费凭据。采纳核对结果随动作账本写入,不写另一份消费快照。 + provenance = dict(check['provenance']) + provenance['consumption_id'] = hard.get('consumption_id') + return {'ok': True, 'proposal': dict(p, hard_numbers=dict(hard, **provenance))} + 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): @@ -64,7 +278,8 @@ def _failed_recheck(p, out): 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) + 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': '提议已被更新,请刷新后查看当前结果', @@ -87,56 +302,3 @@ def withdraw_recorded_rejections(): (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__}'} diff --git a/app/services/proposal_service.py b/app/services/proposal_service.py index 644bb17..e8d8d12 100644 --- a/app/services/proposal_service.py +++ b/app/services/proposal_service.py @@ -234,17 +234,21 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict: deadline = time.monotonic() + max( 0, param_store.get_int("PMS_JUDGE_TICK_BUDGET_SEC", 150)) - for c in list(scanned["candidates"]) + open_cands: - try: - _rk = _unlocked.get((c.get("ts_code"), c.get("action"))) - if _rk: - c["reask"] = _rk - _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, - deadline=deadline) - except Exception as e: - logger.exception("提议分流失败 %s", c.get("ts_code")) - out["errors"].append(f"{c.get('ts_code')} {c.get('action')}: " - f"{type(e).__name__}: {e}") + for group_index, group in enumerate((list(scanned['candidates']), open_cands)): + if group_index == 1 and not dry_run and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False): + from app.services.plan_recheck import refresh_pending + out['prepared_proposals'] = refresh_pending(now=now, deadline=deadline) + for c in group: + try: + _rk = _unlocked.get((c.get("ts_code"), c.get("action"))) + if _rk: + c["reask"] = _rk + _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, + deadline=deadline) + except Exception as e: + logger.exception("提议分流失败 %s", c.get("ts_code")) + out["errors"].append(f"{c.get('ts_code')} {c.get('action')}: " + f"{type(e).__name__}: {e}") # 技术面转空离场去重集写回 (2026-09-11 工作包三): 本轮真处理过 (执行或入队, 未被闸拒) 的 # 转空离场记进去重集, 下次同一翻空不再重复处理。试算不写。清掉 SAR 已翻多的那步在扫描前做过了。 if not dry_run: @@ -774,17 +778,12 @@ def disposition_snapshot(now=None) -> dict: return res -def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, - *, deadline=None, review_only=False): - code, action, side = c["ts_code"], c["action"], c["side"] - is_open = (action == ae.A_OPEN) +def check_candidate_rules(c, view, params, stock_params, brake_active): + code, action, side = c['ts_code'], c['action'], c['side'] + is_open = action == ae.A_OPEN pos = _pos_of(view, code) - # 新建仓的现价取候选自带的那份。**不能取 pos 的**: 从没持仓过的票在账本里没有行, - # `_pos_of` 回的是空壳、没有 price, 取它会是 0, 规则闸一上来就判 PRICE_MISSING。 - price = float(c.get("price") or 0) if is_open else float(pos.get("price") or 0) - - # ---- 一级: 规则闸 (自主动作受刹车约束, is_command=False) ---- - gate = rule_gate.check( + price = float(c.get('price') or 0) if is_open else float(pos.get('price') or 0) + return rule_gate.check( side=side, action=action, qty=c["qty"], price=price, ctx={"ts_code": code, "position": pos, # 当日行情走 _market_ctx 取到的真实快照。原来这里是 @@ -811,6 +810,19 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, # 不拦截, 见 rule_gate 里那段说明。 "stop_price": stock_params.get(code, {}).get("stop_price"), "is_command": False}}) + + +def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, + *, deadline=None, review_only=False, prepared_verdict=None): + code, action, side = c["ts_code"], c["action"], c["side"] + is_open = (action == ae.A_OPEN) + pos = _pos_of(view, code) + # 新建仓的现价取候选自带的那份。**不能取 pos 的**: 从没持仓过的票在账本里没有行, + # `_pos_of` 回的是空壳、没有 price, 取它会是 0, 规则闸一上来就判 PRICE_MISSING。 + price = float(c.get("price") or 0) if is_open else float(pos.get("price") or 0) + + # ---- 一级: 规则闸 (自主动作受刹车约束, is_command=False) ---- + gate = check_candidate_rules(c, view, params, stock_params, brake_active) if not gate["passed"]: out["rejected"].append({"ts_code": code, "action": action, "by": "rule", "failed": gate["failed"]}) @@ -829,6 +841,12 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, and _sd_mode in ("decline", "full")): verdict = {"verdict": "SKIPPED", "reason": "逻辑存疑,不送研判(系统自决直接放弃)", "degraded": False, "raw": None, "confidence": None} + elif prepared_verdict is not None: + verdict = dict(prepared_verdict) # 后台仅额度变化时复用同研究依据的结果。 + if verdict.get('verdict') == judge.REJECT: + out['rejected'].append({'ts_code': code, 'action': action, 'by': 'judge', + 'failed': [verdict.get('reason') or '决策系统驳回']}) + return elif c.get("judge_required"): if not _judge_budget_left(deadline): # 本轮研判时间用尽 —— 整条跳过, 下一跳重来。 @@ -1438,8 +1456,11 @@ def _make_proposal(c, price, verdict) -> str: # 采纳按钮那边直接抄提议里的股数原样落单 (只有卖出侧会按可卖量重算), 所以不在这里 # 改, 后面就没有第二次机会了。 _apply_advice_tier(c, hn, verdict) - from app.core.plan_revision import fingerprint - hn['proposal_revision'] = fingerprint({'qty': c['qty'], 'hard_numbers': hn}) + if c['action'] == ae.A_OPEN and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False): + from app.services.plan_recheck import prepare_new + hn = prepare_new(hn, c['ts_code'], verdict) + from app.core.plan_revision import confirmation_revision + hn['proposal_revision'] = confirmation_revision(c['qty'], hn, verdict) try: pms_repo.insert_proposal( proposal_id=pid, ts_code=c["ts_code"], action=c["action"], qty=c["qty"], diff --git a/app/web/main.py b/app/web/main.py index fab6e6b..b3782bd 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -476,6 +476,13 @@ def api_proposals(status: str = Query("WAIT_USER"), limit: int = 100, tech_service.attach(rows) except Exception as e: # noqa: BLE001 logger.warning("[提议] 技术面立场挂不上: %s", e) + if param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False): + for row in rows: + h = row.get('hard_numbers') or {} + if row.get('action') == 'OPEN' and h.get('kind') != 'strategy': + preparation = h.get('preparation') or {} + row['accept_ready'] = preparation.get('status') == 'ready' and bool(h.get('consumption_id')) + row['prepare_reason'] = preparation.get('reason') or '建议正在准备,完成后可采纳' return rows return ok(_list) @@ -485,8 +492,8 @@ def api_decide_reasons(): """裁决预设理由的两张码表 (采纳 / 驳回), 页面据此铺按钮。台账 015。""" return {"ok": True, "accept": decide_reasons.ACCEPT_REASONS, "decline": decide_reasons.DECLINE_REASONS, - # 浏览器等待要覆盖研判超时与取数,不能先报失败、后端还在处理采纳。 - "request_timeout_ms": max(120, param_store.get_int('PMS_JUDGE_TIMEOUT', 90) + 60) * 1000} + # 采纳只做快速检查,不等待完整研判;超时仍按结果未知处理。 + "request_timeout_ms": 30000} @app.post("/api/proposals/{proposal_id}/decide") diff --git a/app/web/static/index.html b/app/web/static/index.html index 76e7f91..df65023 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -1229,7 +1229,11 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;} · {{ proposals.length }} 条{{ sectClosed.prop?'▸':'▾' }}