Level2 量化实战（二）：逐笔聚合主力净流入，并用日频资金流接口交叉验证对倒

# -*- coding: utf-8 -*-
"""Level-2 主力资金流向：逐笔成交 3 秒聚合 + 价资背离检测"""
import os, socket, time, threading
import pandas as pd

L2_USER = os.getenv("L2_USER", "你的Level2用户名")
L2_PWD  = os.getenv("L2_PWD",  "你的Level2密码")
L2_HOST = os.getenv("L2_TCP_HOST", "实时服务器地址StockApi找管理)")

def parse_tran(body):
    """字段顺序见第 27 篇字段对照表；深市撤单（价格为 0）剔除"""
    p = body.split(",")
    if len(p) < 10 or p[6] == "0":
        return None
    return {"code":  p[2],
            "time":  p[4],
            "price": int(p[6]) / 10000,
            "amt":   int(p[8]) / 10000,
            "dir":   int(p[9])}          # 1=主动买 2=主动卖

class TranStream:
    def __init__(self, symbols, port=18105):
        self.symbols, self.port = symbols, port
        self.records, self._buf = [], ""

    def start(self):
        s = socket.create_connection((L2_HOST, self.port))
        s.sendall(f"DL,{L2_USER},{L2_PWD}".encode())
        time.sleep(1)
        s.sendall((f"DY2,{L2_USER},{L2_PWD},"
                   + ",".join(self.symbols)).encode())
        threading.Thread(target=self._listen, args=(s,), daemon=True).start()

    def _listen(self, s):
        while True:
            self._buf += s.recv(65536).decode(errors="ignore")
            while "<" in self._buf and ">" in self._buf:
                i, j = self._buf.index("<"), self._buf.index(">")
                body, self._buf = self._buf[i+1:j], self._buf[j+1:]
                rec = parse_tran(body)
                if rec:
                    self.records.append(rec)

def t2sec(t):
    return int(t[:2]) * 3600 + int(t[2:4]) * 60 + int(t[4:6])

# ---------- 分层与聚合 ----------
def tier(amt):
    if amt >= 1_000_000: return "main"      # 超大单
    if amt >=   200_000: return "main"      # 大单 → 与超大单合称主力
    if amt >=    40_000: return "mid"
    return "retail"

class FlowMonitor:
    def __init__(self, tick=3):
        self.tick = tick
        self.acc = {}          # code -> dict(main_buy, main_sell, all_buy, all_sell, last_price)

    def run(self, stream):
        buf = []
        while True:
            time.sleep(self.tick)
            buf += stream.records; stream.records = []
            if not buf:
                continue
            for r in buf:
                a = self.acc.setdefault(
                    r["code"], dict(mb=0, ms=0, ab=0, as_=0, px=r["price"]))
                side = "b" if r["dir"] == 1 else "s"
                a["a" + ("b" if side == "b" else "s_")] += r["amt"]
                if tier(r["amt"]) == "main":
                    a["m" + ("b" if side == "b" else "s")] += r["amt"]
                a["px"] = r["price"]
            buf = []
            for code, a in self.acc.items():
                main_net = a["mb"] - a["ms"]
                all_net  = a["ab"] - a["as_"]
                total    = a["ab"] + a["as_"]
                ratio    = main_net / total if total else 0
                print(f"{code} 最新 {a['px']:.2f} | 主力净 {main_net/1e4:>8.0f} 万"
                      f" | 主力净占比 {ratio:>6.1%} | 全口径净 {all_net/1e4:.0f} 万")
            # 价资背离检测：最近 30 分钟价格新高但主力净额走低
            self.divergence()

    def divergence(self):
        # 简化示例：实盘应保存每 tick 的 (px, main_net) 序列做滚动对比
        pass

if __name__ == "__main__":
    stream = TranStream(["000001.SZ", "600000.SH"])
    stream.start()
    FlowMonitor().run(stream)

# -*- coding: utf-8 -*-
"""全量逐笔回放：验证实时聚合口径，并与日频资金流接口对照"""
import gzip, re
import requests
import pandas as pd

HTTP_URL = "http://down.l2api.cn:9090"
TOKEN    = os.getenv("STOCKAPI_TOKEN", "你的StockApiToken")

def download_full_tran(stock, date_str=""):
    """HTTP 下载通道：Login 换取 __beetlex_token（Session Cookie 自动携带），
    GetData 按 offset 翻页取 gzip 压缩的逐笔成交 CSV。"""
    s = requests.Session()
    s.post(f"{HTTP_URL}/Login",
           json={"userName": os.getenv("L2_USER"),
                 "userPwd":  os.getenv("L2_PWD")})
    rows, offset = [], 0
    while True:
        d = s.post(f"{HTTP_URL}/GetData",
                   json={"stock": stock, "type": "Tran",
                         "offset": offset, "gzip": True}).json()
        if not d.get("data"):
            break
        raw = d["data"] if isinstance(d["data"], bytes) \
              else bytes(d["data"], "latin1")
        for line in gzip.decompress(raw).decode().strip().splitlines():
            p = line.strip("<>").split(",")
            if len(p) >= 10 and p[6] != "0":
                rows.append({"time": p[4], "price": int(p[6]) / 10000,
                             "amt": int(p[8]) / 10000, "dir": int(p[9])})
        offset += 1
    return pd.DataFrame(rows)

def replay_flow(df):
    """按 1 分钟聚合，输出主力净流入曲线"""
    df["min"] = df["time"].str[:4]
    df["main"] = df["amt"].apply(lambda a: a >= 200_000)
    g = df.groupby(["min"]).apply(
        lambda x: pd.Series({
            "close":     x["price"].iloc[-1],
            "main_net":  x.loc[x["main"] & (x["dir"] == 1), "amt"].sum()
                       - x.loc[x["main"] & (x["dir"] == 2), "amt"].sum(),
            "turnover":  x["amt"].sum()}))
    g["main_ratio"] = g["main_net"] / g["turnover"]
    return g

def daily_flow_compare(code, start, end):
    """日频资金流接口（复用 REST 通道）做交叉验证：
    逐笔主力净额 vs 接口主力净额，差值大 = 当日存在对倒/拆单。"""
    d = requests.get("https://www.stockapi.com.cn/v1/base/codeFlowHistory",
                     params={"token": TOKEN, "code": code,
                             "startDate": start, "endDate": end,
                             "pageNo": 1, "pageSize": 200}).json()
    if d.get("code") != 20000:
        return None
    rec = pd.DataFrame(d["data"]["records"])
    rec["mainNetTurnover"] = pd.to_numeric(rec["mainNetTurnover"], errors="coerce")
    return rec[["time", "mainNetTurnover"]]

if __name__ == "__main__":
    df = download_full_tran("000001.SZ")
    curve = replay_flow(df)
    print(curve.tail(30).to_string())
    # 与日频口径对照
    print(daily_flow_compare("000001", "2026-10-20", "2026-10-28"))