fanhuadesenlinnn/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis synced 2026-09-20 10:53:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix missing capital flow buy guard (#1289)

Browse Source
This commit is contained in:
mumu
2026-05-15 20:58:56 +08:00
committed by GitHub
parent 1a64da4700
commit e22cf8e1d8
3 changed files with 184 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
docs/CHANGELOG.md
View File

@@ -42,6 +42,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
- [修复] Docker 启动入口自动修复 `data` / `logs` / `reports` 挂载目录权限并降权运行，文档化的 Compose `exec` 手动命令显式使用 `dsa` 用户，避免普通部署需要手动 `chown` / `chmod`。
- [修复] Web 首页大盘复盘结果改由主内容滚动区承载，避免 loading 切换到长结果后下方报告区域被截断或无法继续滚动。
- [修复] Web 设置页为通知测试与 Agent/通知配置区域增加局部运行时错误兜底，异常时提示提供 Windows 桌面端 `desktop.log`，避免整页黑屏。
- [修复] 资金流数据不可用时将直接买入结论降级为观察，避免缺失数据被误读为高置信买入依据。
- [修复] 调高基本面聚合默认超时预算，降低 Windows/Docker 环境下整段基本面 timeout 的概率。
## [3.16.0] - 2026-05-10

191
src/analyzer.py
View File

@@ -721,23 +721,38 @@ def stabilize_decision_with_structure(
price_position.get("resistance_level"),
_first_list_value(trend_dict.get("resistance_levels")),
)
flow_bias, flow_reason = _capital_flow_bias_with_status(fundamental_context)
if flow_bias == "unavailable":
if isinstance(fundamental_context, dict) and "capital_flow" in fundamental_context:
_set_decision_stability_unavailable(
result,
language,
current_price=current_price,
support=support,
resistance=resistance,
flow_status=flow_reason,
)
return
decision_type = infer_decision_type_from_advice(
getattr(result, "decision_type", ""),
default=getattr(result, "decision_type", "hold") or "hold",
)
decision_type = decision_type if decision_type in {"buy", "hold", "sell"} else "hold"
advice_decision_type = infer_decision_type_from_advice(
getattr(result, "operation_advice", ""),
default="",
)
flow_bias, flow_reason = _capital_flow_bias_with_status(fundamental_context)
if flow_bias == "unavailable":
if isinstance(fundamental_context, dict) and "capital_flow" in fundamental_context:
if decision_type == "buy" or advice_decision_type == "buy":
_downgrade_buy_without_capital_flow(
result,
language,
current_price=current_price,
support=support,
resistance=resistance,
flow_status=flow_reason,
)
else:
_set_decision_stability_unavailable(
result,
language,
current_price=current_price,
support=support,
resistance=resistance,
flow_status=flow_reason,
)
return
if current_price is None:
return
@@ -1019,6 +1034,107 @@ def _set_decision_stability_unavailable(
_sync_stability_dashboard_fields(result)
def _bound_hold_watch_sentiment_score(result: "AnalysisResult") -> None:
try:
score = int(getattr(result, "sentiment_score", 50))
except (TypeError, ValueError):
score = 50
result.sentiment_score = min(59, max(45, score))
def _apply_hold_watch_dashboard(
result: "AnalysisResult",
language: str,
*,
advice: str,
reason: str,
current_price: Optional[float],
support: Optional[float],
resistance: Optional[float],
flow_bias: str,
no_position: str,
has_position: str,
capital_flow_status: Optional[str] = None,
) -> None:
result.operation_advice = advice
dashboard = result.dashboard if isinstance(result.dashboard, dict) else {}
result.dashboard = dashboard
core = dashboard.get("core_conclusion")
if not isinstance(core, dict):
core = {}
dashboard["core_conclusion"] = core
core["signal_type"] = "🟡持有观望" if language == "zh" else "🟡 Hold / Watch"
core["one_sentence"] = f"{advice}：{reason}" if language == "zh" else f"{advice}: {reason}"
position_advice = core.get("position_advice")
if not isinstance(position_advice, dict):
position_advice = {}
core["position_advice"] = position_advice
position_advice["no_position"] = no_position
position_advice["has_position"] = has_position
stability = {
"applied": True,
"reason": reason,
"current_price": current_price,
"support": support,
"resistance": resistance,
"capital_flow_bias": flow_bias,
}
if capital_flow_status is not None:
stability["capital_flow_status"] = capital_flow_status
dashboard["decision_stability"] = stability
if reason and reason not in str(result.risk_warning or ""):
sep = "；" if language == "zh" else "; "
result.risk_warning = f"{result.risk_warning}{sep}{reason}" if result.risk_warning else reason
result.buy_reason = reason or result.buy_reason
def _downgrade_buy_without_capital_flow(
result: "AnalysisResult",
language: str,
*,
current_price: Optional[float],
support: Optional[float],
resistance: Optional[float],
flow_status: str,
) -> None:
status_text = _capital_flow_status_for_stability(flow_status, language)
if language == "zh":
advice = "持有观察"
reason = f"{status_text}，买入结论缺少资金面确认，先按观察处理。"
no_position = "空仓先不追买，等待资金流恢复、支撑确认或有效突破后再行动。"
has_position = "持仓以关键支撑为风控线，资金流恢复前控制仓位。"
confidence = "低"
else:
advice = "Hold and watch"
reason = f"{status_text}; the buy call lacks capital-flow confirmation, so treat it as watch-only."
no_position = "Do not chase; wait for capital-flow recovery, support confirmation, or a valid breakout."
has_position = "Use key support as the risk line and keep position size controlled until capital flow recovers."
confidence = "Low"
result.decision_type = "hold"
result.confidence_level = confidence
_bound_hold_watch_sentiment_score(result)
_apply_hold_watch_dashboard(
result,
language,
advice=advice,
reason=reason,
current_price=current_price,
support=support,
resistance=resistance,
flow_bias="unavailable",
no_position=no_position,
has_position=has_position,
capital_flow_status=status_text,
)
_sync_stability_dashboard_fields(result)
logger.info("[decision_stability] Downgraded buy because capital flow is unavailable: %s", flow_status)
def _downgrade_to_structural_hold(
result: "AnalysisResult",
language: str,
@@ -1031,11 +1147,7 @@ def _downgrade_to_structural_hold(
flow_bias: str,
) -> None:
result.decision_type = "hold"
try:
score = int(getattr(result, "sentiment_score", 50))
except (TypeError, ValueError):
score = 50
result.sentiment_score = min(59, max(45, score))
_bound_hold_watch_sentiment_score(result)
_set_structural_hold_wording(
result,
language,
@@ -1096,37 +1208,24 @@ def _set_structural_hold_wording(
elif language == "en" and advice_key == "range":
result.trend_prediction = "Sideways"
dashboard = result.dashboard if isinstance(result.dashboard, dict) else {}
result.dashboard = dashboard
core = dashboard.get("core_conclusion")
if not isinstance(core, dict):
core = {}
dashboard["core_conclusion"] = core
core["signal_type"] = "🟡持有观望" if language == "zh" else "🟡 Hold / Watch"
core["one_sentence"] = f"{advice}：{reason}" if language == "zh" else f"{advice}: {reason}"
position_advice = core.get("position_advice")
if not isinstance(position_advice, dict):
position_advice = {}
core["position_advice"] = position_advice
if language == "zh":
position_advice["no_position"] = "空仓先不追涨杀跌，等待支撑确认、放量突破或资金回流后再行动。"
position_advice["has_position"] = "持仓以关键支撑为风控线，未跌破前以观察和分批控仓为主。"
no_position = "空仓先不追涨杀跌，等待支撑确认、放量突破或资金回流后再行动。"
has_position = "持仓以关键支撑为风控线，未跌破前以观察和分批控仓为主。"
else:
position_advice["no_position"] = "Do not chase or panic; wait for support confirmation, breakout, or renewed inflow."
position_advice["has_position"] = "Use key support as the risk line and manage position size unless support fails."
dashboard["decision_stability"] = {
"applied": True,
"reason": reason,
"current_price": current_price,
"support": support,
"resistance": resistance,
"capital_flow_bias": flow_bias,
}
if reason and reason not in str(result.risk_warning or ""):
sep = "；" if language == "zh" else "; "
result.risk_warning = f"{result.risk_warning}{sep}{reason}" if result.risk_warning else reason
result.buy_reason = reason or result.buy_reason
no_position = "Do not chase or panic; wait for support confirmation, breakout, or renewed inflow."
has_position = "Use key support as the risk line and manage position size unless support fails."
_apply_hold_watch_dashboard(
result,
language,
advice=advice,
reason=reason,
current_price=current_price,
support=support,
resistance=resistance,
flow_bias=flow_bias,
no_position=no_position,
has_position=has_position,
)
logger.info("[decision_stability] Applied structural hold calibration: %s", reason_key)

52
tests/test_decision_stability.py
View File

@@ -128,7 +128,7 @@ def test_downgrades_buy_mid_range_with_neutral_fund_flow() -> None:
assert "资金流不明确" in result.risk_warning
def test_skips_calibration_when_capital_flow_is_unavailable() -> None:
def test_downgrades_buy_when_capital_flow_is_unavailable() -> None:
buy_result = _result(
decision_type="buy",
operation_advice="买入",
@@ -154,17 +154,20 @@ def test_skips_calibration_when_capital_flow_is_unavailable() -> None:
_unsupported_fund_flow(),
)
assert buy_result.decision_type == "buy"
assert buy_result.operation_advice == "买入"
assert buy_result.dashboard["decision_stability"]["applied"] is False
assert "未使用资金流校准" in buy_result.dashboard["decision_stability"]["reason"]
assert buy_result.decision_type == "hold"
assert buy_result.operation_advice == "持有观察"
assert buy_result.confidence_level == "低"
assert buy_result.sentiment_score <= 59
assert buy_result.dashboard["decision_stability"]["applied"] is True
assert "买入结论缺少资金面确认" in buy_result.dashboard["decision_stability"]["reason"]
assert buy_result.dashboard["core_conclusion"]["signal_type"] == "🟡持有观望"
assert sell_result.decision_type == "sell"
assert sell_result.operation_advice == "卖出"
assert sell_result.dashboard["decision_stability"]["applied"] is False
assert "未使用资金流校准" in sell_result.dashboard["decision_stability"]["reason"]
def test_skips_calibration_when_capital_flow_values_are_na() -> None:
def test_downgrades_buy_when_capital_flow_values_are_na() -> None:
result = _result(
decision_type="buy",
operation_advice="买入",
@@ -189,13 +192,34 @@ def test_skips_calibration_when_capital_flow_values_are_na() -> None:
},
)
assert result.decision_type == "buy"
assert result.operation_advice == "买入"
assert result.dashboard["decision_stability"]["applied"] is False
assert "未使用资金流校准" in result.dashboard["decision_stability"]["reason"]
assert result.decision_type == "hold"
assert result.operation_advice == "持有观察"
assert result.dashboard["decision_stability"]["applied"] is True
assert "资金流数据缺失" in result.dashboard["decision_stability"]["capital_flow_status"]
def test_skips_calibration_when_capital_flow_status_is_unavailable_case_insensitive() -> None:
def test_downgrades_buy_advice_when_decision_type_is_hold_and_capital_flow_unavailable() -> None:
result = _result(
decision_type="hold",
operation_advice="建议买入",
score=68,
current_price=32.0,
)
stabilize_decision_with_structure(
result,
SimpleNamespace(support_levels=[30.0], resistance_levels=[34.0]),
_unsupported_fund_flow(),
)
assert result.decision_type == "hold"
assert result.operation_advice == "持有观察"
assert result.sentiment_score <= 59
assert result.dashboard["decision_stability"]["applied"] is True
assert "买入结论缺少资金面确认" in result.dashboard["decision_stability"]["reason"]
def test_downgrades_buy_when_capital_flow_status_is_unavailable_case_insensitive() -> None:
buy_result = _result(
decision_type="buy",
operation_advice="买入",
@@ -209,9 +233,9 @@ def test_skips_calibration_when_capital_flow_status_is_unavailable_case_insensit
_unsupported_fund_flow_caps(),
)
assert buy_result.decision_type == "buy"
assert buy_result.operation_advice == "买入"
assert buy_result.dashboard["decision_stability"]["applied"] is False
assert buy_result.decision_type == "hold"
assert buy_result.operation_advice == "持有观察"
assert buy_result.dashboard["decision_stability"]["applied"] is True
assert "暂不支持" in str(buy_result.dashboard["decision_stability"]["capital_flow_status"])