tradingSystem/app/core/consensus.py

92 lines
4.8 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding: utf-8 -*-
"""三源合议与路由 (2026-09-11, 纯逻辑, 零外部依赖)。
三票 (基本面 / 技术面 / 择时系统) 各是看多/看空/中性/无读数之一, 合成方向再定路由。
方案第四节的合议规则 (台账 009):
三票多数决定方向; 无读数弃权不计; 平局中性; 只有一方表态算弱; 三方一致算强。
路由表见方案附录丙。方向词与合议词一律中文原文, 取值只有看多/看空/中性/无读数。
"""
from __future__ import annotations
_BULL, _BEAR, _NEUT, _NA = "看多", "看空", "中性", "无读数"
# 方向看多但择时看空时的两种去向 (2026-09-14 系统自决包): 默认交人; 自决开着时改回观察等择时转多。
_R_WAIT_TIMING = "等择时转多"
def _reason(fund, tech, timing, direction, strength):
head = "".join([f"基本面{fund}", f"技术面{tech}", f"择时{timing}"])
return head + "" + direction + (("·" + strength) if strength else "")
def _route(fund, tech, timing, direction, tech_phase, fund_required, weak_confirm=True,
timing_dissent="confirm"):
"""附录丙路由表, 从上到下第一个命中。返回 (路由, 原因)。路由取值: 放行/观察/交人/跳过。"""
if fund_required and fund == _NA:
return "跳过", "没有买方评析"
if direction == _BEAR:
return "跳过", f"方向看空(基本面{fund}、技术面{tech}、择时{timing}"
if direction == _NEUT and tech_phase == "收口等待" and fund != _BEAR:
return "观察", "等技术面开口"
if direction == _NEUT and fund == _BULL and tech == _BEAR:
return "观察", "等技术面转向"
if fund == _BEAR and tech == _BULL:
return "交人", "基本面看空、技术面看多,试探仓"
if direction == _BULL:
voiced = [v for v in (fund, tech, timing) if v in (_BULL, _BEAR)]
if timing == _BEAR: # 异议在择时系统
# 默认交人 (逐字回旧); 系统自决开着时改回观察, 等择时转多再走放行 (2026-09-14)。
if timing_dissent == "observe":
return "观察", _R_WAIT_TIMING
return "交人", "方向看多但择时系统有异议"
if len(voiced) <= 1: # 只有一方表态
# 弱表态: weak_confirm 开着交人, 关掉放行 (2026-09-14 加固, 登记 PMS_CONSENSUS_WEAK_CONFIRM)
if weak_confirm:
return "交人", "方向看多但只有一方表态"
return "放行", "方向看多但只有一方表态(弱表态放行)"
return "放行", "方向看多" # 看多强, 或异议只在技术面
if fund == _NA and tech == _NA and timing == _NA:
return "放行", "三方都无读数,机械方案"
return "观察", "方向中性"
def decide(fund, tech, timing, *, tech_phase=None, fund_required=True, weak_confirm=True,
timing_dissent="confirm") -> dict:
"""fund/tech/timing 是三个立场字符串 (看多/看空/中性/无读数)。tech_phase 是技术面相位
(收口等待 等), 路由用。fund_required: 无买方评析是否拦 (开关 PMS_FUND_REQUIRED)。
weak_confirm: 方向看多但只有一方表态时是否交人 (开关 PMS_CONSENSUS_WEAK_CONFIRM, 默认真;
关掉则放行)。timing_dissent: 方向看多但择时看空时的去向, confirm 交人 (默认, 逐字回旧) /
observe 改观察等择时转多 (系统自决开着时传, 观察结果带 disp=wait_timing)。"""
votes = {"fund": fund, "tech": tech, "timing": timing}
voiced = [v for v in (fund, tech, timing) if v in (_BULL, _BEAR)]
bulls = sum(1 for v in voiced if v == _BULL)
bears = sum(1 for v in voiced if v == _BEAR)
# ---- 方向: 多数决, 无读数弃权不计, 平局中性 ----
if bulls > bears:
direction = _BULL
elif bears > bulls:
direction = _BEAR
else:
direction = _NEUT
# ---- 强弱: 只有一方表态弱, 三方一致强, 其余 (两方表态) 为中 ----
if direction == _NEUT:
strength = None
elif len(voiced) <= 1:
strength = ""
elif bulls == 3 or bears == 3:
strength = ""
else:
strength = ""
route, route_reason = _route(fund, tech, timing, direction, tech_phase, fund_required,
weak_confirm, timing_dissent)
out = {"direction": direction, "votes": votes, "strength": strength,
"reason": _reason(fund, tech, timing, direction, strength),
"route": route, "route_reason": route_reason}
# 择时异议改观察这一种, 处置词记 wait_timing (页面显示「择时待转多」; 动作引擎观察分支据它选标签)。
if route == "观察" and route_reason == _R_WAIT_TIMING:
out["disp"] = "wait_timing"
return out