From fca5bfc24d1284c43ab9bb1d59d4d7a0afb59a2c Mon Sep 17 00:00:00 2001 From: mike Date: Thu, 23 Jul 2026 10:33:47 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=82=A1=E7=A5=A8=E5=88=86=E6=9E=90?= =?UTF-8?q?=E5=8A=A0=E6=94=BF=E7=AD=96=E6=B3=A8=E5=85=A5(A=E8=82=A1CN/?= =?UTF-8?q?=E6=B8=AF=E7=BE=8E=E8=82=A1CN+INTL)=20+=20=E6=96=B0=E5=BB=BAcal?= =?UTF-8?q?c=5Fcn=5Flevels.py+fetch=5Fpolicy.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- strategy-management/scripts/calc_cn_levels.py | 34 ++++- strategy-management/scripts/calc_hk_levels.py | 24 +++- strategy-management/scripts/calc_us_levels.py | 24 +++- strategy-management/scripts/fetch_policy.py | 127 ++++++++++++++++++ 4 files changed, 203 insertions(+), 6 deletions(-) create mode 100644 strategy-management/scripts/fetch_policy.py diff --git a/strategy-management/scripts/calc_cn_levels.py b/strategy-management/scripts/calc_cn_levels.py index 0b3ebe5..22c213c 100644 --- a/strategy-management/scripts/calc_cn_levels.py +++ b/strategy-management/scripts/calc_cn_levels.py @@ -121,6 +121,26 @@ def calc_levels(symbol: str, klines: list, quote: dict, side: str, min_rr: float ), strategy +def _read_policy(path): + """读取政策文件，不存在则返回 None""" + p = Path(path) + return p.read_text(encoding="utf-8").strip() if p.exists() else None + + +def _build_header(args, strategy_label): + """构建推送头部: 政策 + 策略""" + lines = [] + # 政策注入 + if args.policy: + pol = _read_policy(args.policy) + if pol: + lines.append(pol) + # 策略说明 + if strategy_label and strategy_label != 'rsi2_revert': + lines.append(f"(策略: {strategy_label})") + return "\n".join(lines) if lines else "" + + def format_cn_output(levels, change_pct, current_price, side, symbol, info, mode): """A 股格式: ¥ 符号, 股息率""" name = info.get('name', symbol) @@ -164,6 +184,9 @@ def main(): ap.add_argument('--strategy', default='rsi2_revert', choices=['rsi2_revert', 'vwap_revert', 'early_bird', 'turtle_breakout'], help='策略 (default rsi2_revert, A 股推荐)') + ap.add_argument('--policy', + default='/tmp/policy_cn.txt', + help='国内政策文件路径 (default /tmp/policy_cn.txt, 空则跳过)') args = ap.parse_args() # 过滤高股息 @@ -229,9 +252,16 @@ def main(): print() if output_lines: - header = f"📊 A 股日内做T点位 ({today})\n⚠️ 仅参考, 不交易\n" + header_parts = [] + pol = _read_policy(args.policy) + if pol: + header_parts.append(pol) + header_parts.append(f"📊 A 股日内做T点位 ({today})\n⚠️ 仅参考, 不交易") + if args.strategy != 'rsi2_revert': + header_parts.append(f"(策略: {args.strategy})") + header = "\n".join(header_parts) print("\n=== QQ 推送内容 ===") - print(header + "\n---\n".join(output_lines)) + print(header + "\n" + "\n---\n".join(output_lines)) else: print("\n💤 全部场景否决, 无输出") diff --git a/strategy-management/scripts/calc_hk_levels.py b/strategy-management/scripts/calc_hk_levels.py index 74d0eab..b23a66d 100644 --- a/strategy-management/scripts/calc_hk_levels.py +++ b/strategy-management/scripts/calc_hk_levels.py @@ -148,6 +148,11 @@ def calc_levels(symbol: str, klines: list, quote: dict, side: str, min_rr: float ), strategy +def _read_policy(path): + p = Path(path) + return p.read_text(encoding="utf-8").strip() if p.exists() else None + + def format_qq_output(levels, change, current_price, side, symbol, score, adr, mode): """格式化为 QQ 推送文本 (单条)""" if mode == 'strict': @@ -186,6 +191,10 @@ def main(): ap.add_argument('--strategy', default='rsi2_revert', choices=['rsi2_revert', 'vwap_revert', 'early_bird', 'turtle_breakout'], help='策略 (default rsi2_revert)') + ap.add_argument('--policy-cn', default='/tmp/policy_cn.txt', + help='国内政策文件 (空/不存在则跳过)') + ap.add_argument('--policy-intl', default='/tmp/policy_intl.txt', + help='国际政策文件 (空/不存在则跳过)') args = ap.parse_args() if not os.path.exists(CANDIDATE_FILE): @@ -262,9 +271,20 @@ def main(): print() if output_lines: - header = f"📊 港股日内做T点位 ({date})\n⚠️ 仅参考, 不交易\n" + header_parts = [] + pol_cn = _read_policy(args.policy_cn) + pol_intl = _read_policy(args.policy_intl) + if pol_cn: + header_parts.append(pol_cn) + if pol_intl: + header_parts.append(pol_intl) + header_parts.append(f"📊 港股日内做T点位 ({date})") + if args.strategy != 'rsi2_revert': + header_parts.append(f"(策略: {args.strategy})") + header_parts.append("⚠️ 仅参考, 不交易") + header = "\n".join(header_parts) print("\n=== QQ 推送内容 ===") - print(header + "\n---\n".join(output_lines)) + print(header + "\n" + "\n---\n".join(output_lines)) else: print("\n💤 全部场景否决, 无输出") diff --git a/strategy-management/scripts/calc_us_levels.py b/strategy-management/scripts/calc_us_levels.py index d26a07a..32892e4 100644 --- a/strategy-management/scripts/calc_us_levels.py +++ b/strategy-management/scripts/calc_us_levels.py @@ -136,6 +136,11 @@ def calc_levels(symbol: str, klines: list, quote: dict, side: str, min_rr: float ), strategy +def _read_policy(path): + p = Path(path) + return p.read_text(encoding="utf-8").strip() if p.exists() else None + + def format_qq_output(levels, change, current_price, side, symbol, score, adr, mode): if mode == 'strict': return ( @@ -173,6 +178,10 @@ def main(): ap.add_argument('--strategy', default='rsi2_revert', choices=['rsi2_revert', 'vwap_revert', 'early_bird', 'turtle_breakout'], help='策略 (default rsi2_revert)') + ap.add_argument('--policy-cn', default='/tmp/policy_cn.txt', + help='国内政策文件 (空/不存在则跳过)') + ap.add_argument('--policy-intl', default='/tmp/policy_intl.txt', + help='国际政策文件 (空/不存在则跳过)') args = ap.parse_args() if not os.path.exists(CANDIDATE_FILE): @@ -246,9 +255,20 @@ def main(): print() if output_lines: - header = f"📊 美股日内做T点位 ({date})\n⚠️ 仅参考, 不交易\n" + header_parts = [] + pol_cn = _read_policy(args.policy_cn) + pol_intl = _read_policy(args.policy_intl) + if pol_cn: + header_parts.append(pol_cn) + if pol_intl: + header_parts.append(pol_intl) + header_parts.append(f"📊 美股日内做T点位 ({date})") + if args.strategy != 'rsi2_revert': + header_parts.append(f"(策略: {args.strategy})") + header_parts.append("⚠️ 仅参考, 不交易") + header = "\n".join(header_parts) print("\n=== QQ 推送内容 ===") - print(header + "\n---\n".join(output_lines)) + print(header + "\n" + "\n---\n".join(output_lines)) else: print("\n💤 全部场景否决, 无输出") diff --git a/strategy-management/scripts/fetch_policy.py b/strategy-management/scripts/fetch_policy.py new file mode 100644 index 0000000..c8b0a46 --- /dev/null +++ b/strategy-management/scripts/fetch_policy.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 +""" +fetch_policy.py - 读取 policy-news-monitor cron 最新 session，提取政策标题摘要 + +用法: + python3 fetch_policy.py --type cn # 国内政策 → /tmp/policy_cn.txt + python3 fetch_policy.py --type intl # 国际政策 → /tmp/policy_intl.txt + python3 fetch_policy.py --type both # 两者都抓 + +集成方式: + cron 先跑 policy-news-monitor 生成 session 文件， + 本脚本读取最新 session 提取标题，供股票分析脚本注入。 +""" +import argparse +import re +import sys +import os +from pathlib import Path +from datetime import datetime, date + +REF_DIR = Path(__file__).parent.parent / "references" +_ref_dir = REF_DIR # 运行时覆盖 + + +def latest_session(pattern): + """返回匹配 pattern 的最新 session 文件路径""" + sessions = sorted( + _ref_dir.glob(f"session-*{pattern}*.md"), + key=lambda p: p.stat().st_mtime, + reverse=True + ) + return sessions[0] if sessions else None + + +def extract_headlines(session_path, max_lines=8): + """从 session 文件提取前 N 条政策新闻标题""" + if not session_path or not session_path.exists(): + return None + text = session_path.read_text(encoding="utf-8") + lines = [] + # 匹配形如 "· 央行宣布降准" 或 "1. 政策标题" 的行 + for line in text.splitlines(): + line = line.strip() + if not line: + continue + # 跳过代码块、标题、URL + if line.startswith("#") or line.startswith("```") or line.startswith("http"): + continue + # 提取列表项和政策标题 + m = re.match(r"^[\d\.\、\-\•\◉]+\s*[\[【]?\s*([^\]\n]{8,60})", line) + if m: + title = m.group(1).strip() + if "免责" not in title and len(title) > 6: + lines.append(f" · {title}") + elif len(line) > 8 and len(line) < 80 and "：" in line: + # "央行：降准" 格式 + title = line.strip() + if "免责" not in title: + lines.append(f" · {title}") + if len(lines) >= max_lines: + break + return "\n".join(lines) if lines else None + + +def fetch_cn(): + """国内政策: 读最新 domestic session""" + session = latest_session("domestic") + if not session: + return None, "无 domestic session" + headlines = extract_headlines(session) + if not headlines: + return None, f"提取失败: {session.name}" + return headlines, session.name + + +def fetch_intl(): + """国际政策: 读最新 international session""" + session = latest_session("international") + if not session: + return None, "无 international session" + headlines = extract_headlines(session) + if not headlines: + return None, f"提取失败: {session.name}" + return headlines, session.name + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument('--type', default='both', choices=['cn', 'intl', 'both']) + ap.add_argument('--ref', default=None, help='references 目录') + args = ap.parse_args() + + # 动态设置 ref dir（允许 cron 指定不同路径） + ref_path = Path(args.ref) if args.ref else REF_DIR + if ref_path.exists(): + _ref_dir = ref_path + else: + _ref_dir = REF_DIR + + date_str = date.today().isoformat() + done = [] + + if args.type in ('cn', 'both'): + headlines, info = fetch_cn() + if headlines: + Path("/tmp/policy_cn.txt").write_text(f"# 国内政策摘要 {date_str}（来源: {info}）\n{headlines}\n") + print(f"✅ 国内政策 → /tmp/policy_cn.txt（{info}）") + done.append("cn") + else: + print(f"⚠️ 国内政策: {info}") + + if args.type in ('intl', 'both'): + headlines, info = fetch_intl() + if headlines: + Path("/tmp/policy_intl.txt").write_text(f"# 国际政策摘要 {date_str}（来源: {info}）\n{headlines}\n") + print(f"✅ 国际政策 → /tmp/policy_intl.txt（{info}）") + done.append("intl") + else: + print(f"⚠️ 国际政策: {info}") + + if not done: + print("⚠️ 警告: 两个都没抓到，exit 1") + sys.exit(1) + + +if __name__ == '__main__': + main()