#!/usr/bin/env python3 """「新高法」选股器 —— 依据金亨俊新高法，落到 A 股日线口径。 规则来源（视频口播整理）： 选股：当日成交额前 50（剔 ETF）→ 创阶段新高（近一年）→ 站上 120 日线 → 进自选不追 买点：等「上升阴线」——上涨途中收阴线（十字星也算），且 ① 拉升段成交额够大 ② 阴线收盘仍在 5 日线之上 ③ 已看清压力线且资金在买 卖出：次日开盘走弱即止损；否则持有到跌破 5 日线；高位放量达上涨段 5 倍视为见顶 纪律：同时只持 2~3 只，仓位事先定好，不追没等到买点的票 A 股权衡（与原策略的差异，必须如实标注）： - 成交额门槛：韩股 500 亿韩元 ≈ 2.5 亿人民币 → A 股取「拉升段单日成交额 ≥ 2.5 亿」 - 「机构外资在买」：A 股无同口径实时数据 → 用「收盘价高于当日均价（买盘主导）」 +「当日成交额 ≥ 20 日均额 1.2 倍」双代理，属降级指标 - 一年新高 → 250 交易日内最高价 用法： python picker.py --out ../stock/data/latest.json --top 5 python picker.py --top 5 --pool 80 --relax # 硬条件不足 5 只时放宽一档 """ from __future__ import annotations import argparse import json import sys import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime from pathlib import Path import pandas as pd HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) from aq import fetch_kline, fetch_quote, fetch_snapshot, kline_cached # noqa: E402 STRATEGY = { "key": "new_high", "name": "新高法", "source": "金亨俊（韩国）新高法 · 视频口播整理", "steps": ["成交额前 50（剔 ETF）", "创 250 日新高", "站上 120 日线", "等上升阴线回踩", "阴线收盘仍在 5 日线上", "拉升段成交额 ≥ 2.5 亿", "尾盘买入 / 跌破 5 日线卖出"], } MAX_WORKERS = 4 # 单核机器：网络等待为主，4 路并发不会压满 CPU KLINE_CACHE = HERE.parent / "cache" / "kline" KLINE_TTL = 1800 # K 线缓存 30 分钟：10 分钟一轮的自动更新里，只有首轮真正拉网络 AMT_FLOOR = 2.5e8 # 拉升段单日成交额门槛（元） MIN_MCAP = 3e9 # 30 亿以下流动性太差，剔除 INDICES = [("000001", "上证指数"), ("399001", "深证成指"), ("399006", "创业板指"), ("000680", "科创综指")] def is_etf(code: str, name: str) -> bool: c, n = str(code), str(name) if n.startswith(("ETF", "LOF")) or "ETF" in n.upper(): return True return c[:3] in ("510", "511", "512", "513", "515", "516", "518", "159", "561", "562", "563", "588", "560", "588") def load_cached(code: str): """统一走 aq.kline_cached（磁盘缓存 30 分钟）。""" return kline_cached(code) def analyze(row: dict) -> dict | None: """对单只候选做日线形态判定，返回评分与理由。""" code = row["code"] k = load_cached(code) if k is None or len(k) < 130: return None k = k.reset_index(drop=True) c, v = k["close"], k["volume"] a = k["amount"] amt_estimated = bool(pd.isna(a).all()) if amt_estimated: px = (k["high"] + k["low"] + k["close"]) / 3 a = v * 100 * px # 手 -> 股 × 均价（元） last = k.iloc[-1] prev_close = float(c.iloc[-2]) ma5 = float(c.tail(5).mean()) ma20 = float(c.tail(20).mean()) ma120 = float(c.tail(120).mean()) ma20_prev = float(c.iloc[-25:-5].mean()) hi250_prev = float(c.iloc[-251:-1].max()) if len(c) >= 251 else float(c.iloc[:-1].max()) close = float(last["close"]) open_ = float(last["open"]) body = abs(close - open_) / prev_close is_yin = close < open_ is_doji = body <= 0.004 pattern_ok = (is_yin or is_doji) and close > ma20 and close > ma120 above_ma5 = close > ma5 new_high = close >= hi250_prev * 0.995 near_high = close >= hi250_prev * 0.96 trend_up = close > ma120 and ma20 > ma20_prev # 成交额：已在上方统一成 Series（真实 amount 或 量×均价 估算） amt = float(a.iloc[-1]) amt_20 = float(a.tail(20).mean()) amt_peak10 = float(a.tail(10).max()) vol_ratio = amt / amt_20 if amt_20 else 1.0 push_strong = amt_peak10 >= AMT_FLOOR vwap = (amt / (float(last["volume"]) * 100)) if float(last["volume"]) > 0 else close buy_side = close >= vwap ret5 = (close / float(c.iloc[-6]) - 1) * 100 if len(c) >= 6 else 0.0 ret20 = (close / float(c.iloc[-21]) - 1) * 100 if len(c) >= 21 else 0.0 blowoff = vol_ratio >= 5.0 overheat = ret5 > 20 or ret20 > 50 # 已经飞掉，等回踩而不是追 # ---- 打分（连续化，避免大量并列）---- score, reasons, misses = 0.0, [], [] if new_high: score += 30 reasons.append(f"收盘创 250 日新高（前高 {hi250_prev:.2f}）") score += min(6.0, max(0.0, (close / hi250_prev - 1) * 100 * 1.5)) elif near_high: score += 18 reasons.append(f"逼近 250 日高点，距前高 {(close/hi250_prev-1)*100:+.1f}%") score += min(5.0, max(0.0, 5 - abs(close / hi250_prev - 1) * 100)) else: misses.append(f"距 250 日高点 {(close/hi250_prev-1)*100:+.1f}%，既未创新高也未逼近") if trend_up: score += 18 slope = (ma20 / ma20_prev - 1) * 100 if ma20_prev else 0 score += min(5.0, max(0.0, slope * 1.5)) reasons.append(f"趋势向上：站上 120 日线（{ma120:.2f}），20 日线斜率 {slope:+.2f}%") else: misses.append("未站上 120 日线或 20 日线走平/下行") if pattern_ok: score += 22 reasons.append(("收阴线" if is_yin else "收十字星") + f"，实体 {body*100:.2f}%，仍在 20 日线上方（回踩形态成立）") else: misses.append("当日非「上升阴线/十字星」形态 —— 策略要求等回踩，不追阳线") if above_ma5: score += 14 score += min(4.0, max(0.0, (close / ma5 - 1) * 100)) reasons.append(f"收盘 {close:.2f} 站上 5 日线 {ma5:.2f}") else: misses.append(f"收盘跌破 5 日线 {ma5:.2f}") if push_strong: score += 8 score += min(4.0, amt_peak10 / 1e9) reasons.append(f"拉升段成交额 {amt_peak10/1e8:.2f} 亿（门槛 {AMT_FLOOR/1e8:.1f} 亿）") else: misses.append(f"拉升段成交额不足（{amt_peak10/1e8:.2f} 亿 < {AMT_FLOOR/1e8:.1f} 亿）") if buy_side: score += 6 reasons.append(f"收盘高于当日均价 {vwap:.2f}，买盘主导") else: misses.append("收盘低于当日均价，卖压偏大") if overheat: score -= 30 misses.append(f"短期过热（5 日 {ret5:+.1f}% / 20 日 {ret20:+.1f}%），已飞离买点，等回踩") if blowoff: score -= 25 misses.append(f"当日量能 {vol_ratio:.1f} 倍于 20 日均量，疑似高位放量见顶") # 两段式分档，对应原策略「先入自选，再等回踩买点」： # A = 今日已收上升阴线，具备尾盘买入条件 # B = 各项达标、只差回踩形态 → 入自选等买点 base_ok = (trend_up and above_ma5 and push_strong and (new_high or near_high) and not overheat and not blowoff) if base_ok and pattern_ok: tier = "A" signal = "可买（上升阴线已现）" elif base_ok: tier = "B" signal = "等买点（待回踩收阴）" else: tier = "C" signal = "不符合" return { "code": code, "name": row["name"], "price": round(close, 2), "chg_pct": round(float(last.get("chg_pct", 0)) if pd.notna(last.get("chg_pct", None)) else (close / prev_close - 1) * 100, 2), "amount": round(amt, 0), "score": round(score, 1), "tier": tier, "signal": signal, "reasons": reasons, "misses": misses, "metrics": { "ma5": round(ma5, 2), "ma20": round(ma20, 2), "ma120": round(ma120, 2), "hi250": round(hi250_prev, 2), "dist_to_high_pct": round((close / hi250_prev - 1) * 100, 2), "ret5_pct": round(ret5, 2), "ret20_pct": round(ret20, 2), "vol_ratio": round(vol_ratio, 2), "amount_peak10_yi": round(amt_peak10 / 1e8, 2), "amt_estimated": amt_estimated, "body_pct": round(body * 100, 2), "is_yin": bool(is_yin), }, "plan": { "buy": "下一交易日尾盘仍未破 5 日线时买入（不追高开）", "stop": f"次日开盘走弱立即止损；持有期跌破 5 日线（当前 {ma5:.2f}）卖出", "target": f"不设固定目标，趋势持有；出现 ≥5 倍放量（当前均值 {amt_20/1e8:.2f} 亿）视为见顶", "position": "同时最多持 2~3 只，单只仓位事先定好", }, "spark": [round(float(x), 2) for x in c.tail(60).tolist()], "last_date": str(last["date"]), } def build_market() -> dict: idx = [] for code, name in INDICES: try: q = fetch_quote(code, kind="index") idx.append({"name": name, "price": q.get("price"), "chg_pct": q.get("chg_pct")}) except Exception: # noqa: BLE001 idx.append({"name": name, "price": None, "chg_pct": None}) return {"indices": idx} def pick(top: int = 5, pool: int = 60, ttl: int = 60, relax: bool = False) -> dict: snap = fetch_snapshot("all", ttl) snap = snap[(snap["price"] > 0) & snap["chg_pct"].notna() & (snap["mcap"] >= MIN_MCAP)] snap = snap[~snap["name"].astype(str).str.contains("ST|退", na=False)] snap = snap[~snap.apply(lambda r: is_etf(r["code"], r["name"]), axis=1)] cand = snap.sort_values("amount", ascending=False).head(pool) rows = cand[["code", "name"]].to_dict("records") out = [] with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex: for r in ex.map(analyze, rows): if r: out.append(r) out.sort(key=lambda x: (-x["score"], -x["amount"])) order = {"A": 0, "B": 1, "C": 2} out.sort(key=lambda x: (order[x["tier"]], -x["score"])) picked = [x for x in out if x["tier"] in ("A", "B")][:top] if len(picked) < top and relax: picked += [x for x in out if x["tier"] == "C" and x not in picked][:top - len(picked)] return { "date": datetime.now().strftime("%Y-%m-%d"), "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "trade_date": max((x.get("last_date") or "") for x in out) if out else None, "strategy": STRATEGY, "universe": int(len(snap)), "pool": int(len(cand)), "scanned": len(out), "tierA": sum(1 for x in out if x["tier"] == "A"), "tierB": sum(1 for x in out if x["tier"] == "B"), "picks": picked, "market": build_market(), "notes": [ "A 股无「机构/外资实时买卖」同口径数据，资金条件用「收盘高于均价 + 量能放大」代理，属降级。", "成交额门槛按 500 亿韩元 ≈ 2.5 亿人民币折算。", "涨停股当日多为阳线，天然被「上升阴线」条件过滤，符合原策略不追高的意图。", ], } def main(argv=None): ap = argparse.ArgumentParser(description="新高法选股") ap.add_argument("--out", required=True, help="输出 JSON 路径") ap.add_argument("--top", type=int, default=5) ap.add_argument("--pool", type=int, default=60, help="成交额前 N 只作为候选池") ap.add_argument("--ttl", type=int, default=60) ap.add_argument("--relax", action="store_true", help="A/B 档不足时用 C 档补齐") a = ap.parse_args(argv) d = pick(a.top, a.pool, a.ttl, a.relax) p = Path(a.out) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(d, ensure_ascii=False, indent=2), "utf-8") print(json.dumps({ "saved": str(p), "date": d["date"], "universe": d["universe"], "候选池": d["pool"], "扫描": d["scanned"], "A档": d["tierA"], "B档": d["tierB"], "入选": len(d["picks"]), "picks": [{"code": x["code"], "name": x["name"], "tier": x["tier"], "score": x["score"], "price": x["price"]} for x in d["picks"]], }, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())