tdx-5min-capital-flow-screener

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
5分钟主力资金流强势股筛选 —— 定时执行脚本
口径与 SKILL.md 一致，供 launchd 开盘时段每小时调用：
  1) 资金初筛: 今日主力净流入排名前列 -> 取1分钟资金流K线(累计值)，
     最近5根1分钟差值 = "最近5分钟主力净流入"，排序取前 N
  2) 技术面复核(日线): 收盘价 >= MA5 且 >= MA10；MACD 红柱(柱值>0)；多空线 DKX 前一根为"多"(DKX>其MA10)
  3) 基本面复核: 非ST/退市；PE(TTM)>0；PB<0 淘汰；总市值>=30亿
  4) 输出 Markdown 报告至 results/，休市日自动跳过

数据源（多源降级）:
  - 资金流/排名: 东方财富 push2 → push2delay 镜像（主域名被网络重置时自动切换）
  - 日K线: 腾讯 fqkline（东财日K镜像不可用）
  - 多空指标: 按通达信标准"多空线 DKX"公式本地计算
"""
import json
import os
import sys
import time
import datetime
import urllib.request
import urllib.parse

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))  # skill 根目录
RESULTS_DIR = os.path.join(BASE_DIR, "results")
LOGS_DIR = os.path.join(BASE_DIR, "logs")
LOG_FILE = os.path.join(LOGS_DIR, "runner.log")

TOP_RANK = 60      # 日内主力净流入排名取前 N
TOP_5MIN = 20      # 按 N 分钟主力净流入排序后取前 N 进入技术面复核
MIN_MKTCAP = 3e9   # 总市值门槛 30 亿
# 资金流窗口（分钟）：默认 5；30 分钟变体由 launchd plist 注入环境变量 FLOW_WINDOW_MIN=30
FLOW_WINDOW_MIN = int(os.environ.get("FLOW_WINDOW_MIN", "5"))

RANK_URLS = [
    "https://push2delay.eastmoney.com/api/qt/clist/get",
    "https://push2.eastmoney.com/api/qt/clist/get",
]
FFLOW_URLS = [
    "https://push2delay.eastmoney.com/api/qt/stock/fflow/kline/get",
    "https://push2.eastmoney.com/api/qt/stock/fflow/kline/get",
]
TENCENT_KLINE_URL = "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get"

UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/124.0 Safari/537.36")


def log(msg):
    line = "[%s] %s" % (datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), msg)
    print(line)
    try:
        os.makedirs(LOGS_DIR, exist_ok=True)
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            f.write(line + "\n")
    except OSError:
        pass


def http_get_json(urls, params, tries=2, timeout=12):
    """依次尝试多个镜像域名，每个域名重试 tries 次"""
    if isinstance(urls, str):
        urls = [urls]
    last_err = None
    for url in urls:
        for i in range(tries):
            try:
                qs = urllib.parse.urlencode(params)
                req = urllib.request.Request(
                    url + "?" + qs,
                    headers={"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"})
                with urllib.request.urlopen(req, timeout=timeout) as resp:
                    return json.loads(resp.read().decode("utf-8", "ignore"))
            except Exception as e:  # noqa: BLE001
                last_err = e
                time.sleep(1.0 + 1.5 * i)
    raise RuntimeError("HTTP 请求失败: %s... (%s)" % (urls[0], last_err))


def wan(v):
    """元 -> 万元"""
    try:
        return round(float(v) / 1e4, 1)
    except (TypeError, ValueError):
        return None


def fmt(v):
    return "--" if v is None else str(v)


# ---------------------------------------------------------------- 资金初筛
def get_flow_rank():
    """今日主力净流入排名（沪深A股），返回个股基础信息列表"""
    params = {
        "pn": 1, "pz": TOP_RANK, "po": 1, "np": 1,
        "fltt": 2, "invt": 2, "fid": "f62",
        "fs": "m:0+t:6+f:!2,m:0+t:13+f:!2,m:0+t:80+f:!2,m:1+t:2+f:!2,m:1+t:23+f:!2",
        "fields": "f12,f13,f14,f2,f115,f23,f20,f62",
        "_": int(time.time() * 1000),
    }
    data = http_get_json(RANK_URLS, params)
    diff = (data.get("data") or {}).get("diff") or []
    out = []
    for d in diff:
        out.append({
            "code": str(d.get("f12", "")),
            "market": d.get("f13"),     # 0=深 1=沪
            "name": d.get("f14", ""),
            "pe_ttm": d.get("f115"),    # 市盈率 TTM
            "pb": d.get("f23"),         # 市净率
            "mktcap": d.get("f20"),     # 总市值(元)
            "flow_day": d.get("f62"),   # 今日主力净流入(元)
        })
    return out


def get_fflow_1min(code, market):
    """1分钟资金流K线（累计值）。
    返回 (最近FLOW_WINDOW_MIN分钟主力净流入[元], 今日累计主力净流入[元], 最新时间)"""
    params = {
        "lmt": 0, "klt": 1, "secid": "%s.%s" % (market, code),
        "fields1": "f1,f2,f3,f7",
        "fields2": "f51,f52",
        "_": int(time.time() * 1000),
    }
    data = http_get_json(FFLOW_URLS, params)
    kl = (data.get("data") or {}).get("klines") or []
    if not kl:
        return None, None, None
    bars = [row.split(",") for row in kl]
    today = bars[-1][0].split(" ")[0]
    tb = [b for b in bars if b[0].startswith(today)]
    if not tb:
        return None, None, None
    cums = [float(b[1]) for b in tb]
    last_t = tb[-1][0]
    # 累计值做差: 最近 FLOW_WINDOW_MIN 根1分钟 = 最近N分钟净流入；当日不足N根时基线为0
    base = cums[len(cums) - FLOW_WINDOW_MIN - 1] if len(cums) >= FLOW_WINDOW_MIN + 1 else 0.0
    return cums[-1] - base, cums[-1], last_t


# ---------------------------------------------------------------- 技术指标（日线，腾讯数据源）
def get_daily_kline(code, market, limit=140):
    sym = ("sh" if str(market) == "1" else "sz") + str(code)
    params = {"param": "%s,day,,,%d,qfq" % (sym, limit), "_": int(time.time() * 1000)}
    data = http_get_json(TENCENT_KLINE_URL, params)
    node = (data.get("data") or {}).get(sym) or {}
    rows = node.get("qfqday") or node.get("day") or []
    if not rows:
        raise RuntimeError("腾讯日K为空: %s" % sym)
    # 行格式: [date, open, close, high, low, volume, ...]
    dates = [r[0] for r in rows]
    opens = [float(r[1]) for r in rows]
    closes = [float(r[2]) for r in rows]
    highs = [float(r[3]) for r in rows]
    lows = [float(r[4]) for r in rows]
    return dates, opens, closes, highs, lows


def sma(seq, n):
    if len(seq) < n:
        return None
    return sum(seq[-n:]) / n


def ema(seq, n):
    if not seq:
        return []
    k = 2.0 / (n + 1)
    out = [seq[0]]
    for v in seq[1:]:
        out.append(v * k + out[-1] * (1 - k))
    return out


def macd_hist(closes):
    """TDX 口径: MACD柱 = 2*(DIF-DEA), DEA=EMA(DIF,9)"""
    if len(closes) < 35:
        return None
    dif = [a - b for a, b in zip(ema(closes, 12), ema(closes, 26))]
    dea = ema(dif, 9)
    return 2.0 * (dif[-1] - dea[-1])


def dkx(opens, closes, highs, lows):
    """通达信标准多空线:
       MID := (3*C+L+O+H)/6
       DKX := (20*MID + 19*REF(MID,1) + ... + 1*REF(MID,19)) / 210
       多 = DKX > MA(DKX,10)
    """
    mids = [(3 * c + l + o + h) / 6.0 for c, o, h, l in zip(closes, opens, highs, lows)]
    n = len(mids)
    if n < 30:
        return [], []
    dkx_line = []
    for i in range(n):
        if i < 19:
            dkx_line.append(None)
            continue
        s = 0.0
        for w in range(20):  # w=0..19, 权重 20-w, 引用 REF(mids, w)
            s += (20 - w) * mids[i - w]
        dkx_line.append(s / 210.0)
    ma10 = []
    for i in range(n):
        win = [v for v in dkx_line[max(0, i - 9):i + 1] if v is not None]
        ma10.append(sum(win) / len(win) if len(win) == 10 else None)
    return dkx_line, ma10


# ---------------------------------------------------------------- 主流程
def check_one(stock):
    """返回 (是否全部通过, 明细dict)"""
    detail = dict(stock)
    try:
        last5, day5, t5 = get_fflow_1min(stock["code"], stock["market"])
    except RuntimeError as e:
        return False, {**detail, "fail": "资金流取数失败: %s" % e}
    if last5 is None:
        return False, {**detail, "fail": "无当日资金流数据"}
    detail["flow_5min"], detail["flow_day"], detail["flow_time"] = wan(last5), wan(day5), t5

    try:
        dates, opens, closes, highs, lows = get_daily_kline(stock["code"], stock["market"])
    except RuntimeError as e:
        return False, {**detail, "fail": "K线取数失败: %s" % e}
    if len(closes) < 35:
        return False, {**detail, "fail": "K线数据不足"}
    detail["kline_date"] = dates[-1]

    close = closes[-1]
    ma5, ma10 = sma(closes, 5), sma(closes, 10)
    hist = macd_hist(closes)
    dkx_line, dkx_ma = dkx(opens, closes, highs, lows)
    dkx_prev = dkx_line[-2] if len(dkx_line) >= 2 and dkx_line[-2] is not None else None
    dkxma_prev = dkx_ma[-2] if len(dkx_ma) >= 2 and dkx_ma[-2] is not None else None

    detail.update({
        "close": close,
        "ma5": round(ma5, 2) if ma5 else None,
        "ma10": round(ma10, 2) if ma10 else None,
        "macd": round(hist, 3) if hist is not None else None,
        "dkx_prev": ("多" if dkx_prev > dkxma_prev else "空")
                    if (dkx_prev is not None and dkxma_prev is not None) else "未确认",
    })

    fails = []
    if ma5 is None or ma10 is None or close < ma5 or close < ma10:
        fails.append("未站上5/10日线")
    if hist is None or hist <= 0:
        fails.append("MACD非红柱")
    if detail["dkx_prev"] != "多":
        fails.append("多空前值=%s" % detail["dkx_prev"])

    # 基本面
    name = stock["name"].upper()
    if "ST" in name or "退" in name:
        fails.append("ST/退市风险")
    pe = stock.get("pe_ttm")
    pb = stock.get("pb")
    mcap = stock.get("mktcap")
    if not isinstance(pe, (int, float)) or pe <= 0:
        fails.append("PE(TTM)缺失或为负")
    if isinstance(pb, (int, float)) and pb < 0:
        fails.append("PB为负(资不抵债)")
    if not isinstance(mcap, (int, float)) or mcap < MIN_MKTCAP:
        fails.append("市值不足30亿")

    detail["pe_ttm"] = round(pe, 2) if isinstance(pe, (int, float)) else None
    detail["pb"] = round(pb, 2) if isinstance(pb, (int, float)) else None
    detail["mktcap_yi"] = round(mcap / 1e8, 1) if isinstance(mcap, (int, float)) else None
    detail["fail"] = "；".join(fails)
    return (not fails), detail


def is_trading_day_and_fresh(stocks):
    """最后一根日K日期 == 今天(本机时区) 才视为有效交易时段数据"""
    today = datetime.datetime.now().strftime("%Y-%m-%d")
    for s in stocks[:3]:
        try:
            dates, *_ = get_daily_kline(s["code"], s["market"], limit=5)
            if dates and dates[-1] == today:
                return True
        except RuntimeError:
            continue
    return False


def write_report(passed, near_miss, flow_note):
    os.makedirs(RESULTS_DIR, exist_ok=True)
    now = datetime.datetime.now()
    stamp = now.strftime("%Y%m%d_%H%M%S")
    path = os.path.join(RESULTS_DIR, "screener_%s.md" % stamp)

    lines = []
    lines.append("# %d分钟资金流强势股筛选报告" % FLOW_WINDOW_MIN)
    lines.append("")
    lines.append("- 运行时间: %s" % now.strftime("%Y-%m-%d %H:%M:%S"))
    lines.append("- 数据源: 东方财富(push2delay镜像,排名/资金流) + 腾讯(日K)；TDX MCP 定时任务不可用时的落地实现")
    lines.append("- 资金口径: %s" % flow_note)
    lines.append("- 技术口径: 日线；收盘价≥MA5且≥MA10；MACD柱>0(TDX口径2*(DIF-DEA))；多空线DKX前值>其MA10(前值=多)")
    lines.append("- 基本面口径: 非ST/退市；PE(TTM)>0；PB<0淘汰；总市值≥30亿")
    lines.append("")
    lines.append("## 入选（%d 只）" % len(passed))
    lines.append("")
    if passed:
        lines.append("| 代码 | 名称 | 现价 | %d分钟主力净流入(万) | 今日主力净流入(万) | MA5 | MA10 | MACD柱 | 多空前值 | PE(TTM) | PB | 总市值(亿) |" % FLOW_WINDOW_MIN)
        lines.append("|---|---|---|---|---|---|---|---|---|---|---|---|")
        for d in passed:
            lines.append("| %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s |" % (
                d["code"], d["name"], fmt(d.get("close")),
                fmt(d.get("flow_5min")), fmt(d.get("flow_day")),
                fmt(d.get("ma5")), fmt(d.get("ma10")), fmt(d.get("macd")),
                d.get("dkx_prev", "--"), fmt(d.get("pe_ttm")), fmt(d.get("pb")),
                fmt(d.get("mktcap_yi"))))
    else:
        lines.append("今日无符合条件的股票。")
    lines.append("")
    lines.append("## 最接近名单（未通过原因）")
    lines.append("")
    if near_miss:
        lines.append("| 代码 | 名称 | %d分钟主力净流入(万) | 未通过项 |" % FLOW_WINDOW_MIN)
        lines.append("|---|---|---|---|")
        for d in near_miss[:10]:
            lines.append("| %s | %s | %s | %s |" % (d["code"], d["name"], fmt(d.get("flow_5min")), d.get("fail", "")))
    else:
        lines.append("无")
    lines.append("")
    with open(path, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))
    with open(os.path.join(RESULTS_DIR, "latest.md"), "w", encoding="utf-8") as f:
        f.write("\n".join(lines))
    return path


def main():
    log("开始运行 %d分钟资金流筛选" % FLOW_WINDOW_MIN)
    try:
        rank = get_flow_rank()
    except RuntimeError as e:
        log("FAIL 主力净流入排名取数失败: %s" % e)
        sys.exit(1)
    if not rank:
        log("FAIL 资金流排名为空")
        sys.exit(1)
    log("资金流排名获取 %d 只" % len(rank))

    if not is_trading_day_and_fresh(rank):
        log("SKIP 非交易日/休市，本次不生成报告")
        return

    scored = []
    for s in rank:
        try:
            last5, day5, t5 = get_fflow_1min(s["code"], s["market"])
            if last5 is not None:
                scored.append((last5, day5, t5, s))
        except RuntimeError:
            continue
        time.sleep(0.12)
    if not scored:
        log("FAIL %d分钟资金流数据全部为空" % FLOW_WINDOW_MIN)
        sys.exit(1)
    scored.sort(key=lambda x: x[0], reverse=True)
    top = scored[:TOP_5MIN]
    log("%d分钟排序完成，前 %d 名进入复核（最近时点: %s）" % (FLOW_WINDOW_MIN, len(top), top[0][2]))

    passed, failed = [], []
    for last5, day5, t5, s in top:
        ok, detail = check_one({**s, "_t5": t5})
        (passed if ok else failed).append(detail)
        time.sleep(0.12)
    failed.sort(key=lambda d: (d.get("flow_5min") or 0), reverse=True)

    note = "最近%d根1分钟K线主力净流入合计（时点 %s），今日累计为参考列" % (FLOW_WINDOW_MIN, top[0][2] if top else "--")
    path = write_report(passed, failed, note)
    log("DONE 入选 %d / 复核 %d，报告: %s" % (len(passed), len(top), path))


if __name__ == "__main__":
    main()