一、为什么做这个小工具
刚好有空，后台有粉丝私信能否做一个这样的工具，刚好有空，就做了这样一个工具，我们互相学习交流下。
这个简洁版工具只做一件事：把接口返回的数据转成可以阅读的统计表，方便盘中观察和盘后复盘。
图1：盘口大单分布界面示意
二、这个版本保留哪些功能
图2：程序处理流程
三、完整代码
下面代码保存为：开盘啦委托大单_简洁版.py。先跑通，再考虑接入数据库、前端或实时轮询。
# -*- coding: utf-8 -*-
"""
开盘啦委托大单简洁版
运行：
python 开盘啦委托大单_简洁版.py
修改股票：
把 STOCK_ID = "605100" 改成你要看的 6 位股票代码。
"""
import json
import time
from decimal import Decimal, InvalidOperation
import requests
# ====== 这里改参数 ======
STOCK_ID = "605100" # 股票代码
TUR = 50 # 金额阈值，单位：万元
VOL = 500 # 委托量阈值，单位：手
PAGE_SIZE = 50 # 每次请求条数
START_INDEX = 0 # 从 0 开始拉取
# 如果买卖方向和软件界面相反，把 1 和 2 对调即可
SIDE_MAP = {
"1": "买入",
"2": "卖出",
}
URL = "https://apphq.longhuvip.com/w1/api/index.php"
HEADERS = {
"Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
"Accept": "*/*",
"User-Agent": "lhb/5.12.3 (com.kaipanla.www; build:0; iOS 16.3.1)",
"Accept-Language": "zh-Hans-CN;q=1.0",
}
def to_decimal(value):
"""安全转换数字。"""
if value is None:
return None
try:
return Decimal(str(value).replace(",", "").strip())
except (InvalidOperation, ValueError):
return None
def fetch_data(stock_id, index):
"""请求开盘啦委托大单接口。"""
data = {
"Index": str(index),
"PhoneOSNew": "2",
"StockID": stock_id,
"Tur": str(TUR),
"Type": "3",
"VOrder": "",
"VType": "1",
"VerSion": "5.12.0.3",
"Vol": str(VOL),
"a": "GetWeiTuo_W14",
"apiv": "w34",
"c": "StockL2Data",
"st": str(PAGE_SIZE),
}
resp = requests.post(URL, headers=HEADERS, data=data, timeout=10)
resp.raise_for_status()
return resp.json()
def extract_rows(result):
"""提取返回 JSON 里的 List 明细。"""
rows = result.get("List", [])
return [row for row in rows if isinstance(row, list) and len(row) >= 10]
def parse_row(row):
"""
解析一行委托数据。
row[0] 时间
row[1] 委托编号，带 _CD 表示撤单
row[2] 价格
row[3] 委托量，单位：手
row[4] 委托金额，单位：元
row[5] 买卖方向，默认 1=买入，2=卖出
row[8] 动作标记，0=新增，1=撤单
"""
order_id = str(row[1])
price = to_decimal(row[2])
volume = to_decimal(row[3])
amount_yuan = to_decimal(row[4])
if amount_yuan is not None:
amount_wan = amount_yuan / Decimal("10000")
elif price is not None and volume is not None:
amount_wan = price * volume / Decimal("100")
else:
amount_wan = Decimal("0")
side = SIDE_MAP.get(str(row[5]), "未知")
is_cancel = order_id.endswith("_CD") or str(row[8]) == "1"
return {
"时间": row[0],
"委托编号": order_id,
"价格": price,
"手数": volume,
"金额万": amount_wan,
"方向": side,
"动作": "撤单" if is_cancel else "新增委托",
}
def fetch_all_rows(stock_id):
"""分页拉取全部数据。"""
all_rows = []
seen = set()
index = START_INDEX
while True:
result = fetch_data(stock_id, index)
print(f"\n========== 原始返回 JSON：Index={index} ==========")
print(json.dumps(result, ensure_ascii=False, indent=2))
rows = extract_rows(result)
for row in rows:
key = json.dumps(row, ensure_ascii=False)
if key not in seen:
seen.add(key)
all_rows.append(row)
count = int(result.get("Count", 0) or 0)
end = int(result.get("end", index + len(rows) - 1) or 0)
if not rows:
break
if count and len(all_rows) >= count:
break
if len(rows) < PAGE_SIZE:
break
index = end + 1
time.sleep(0.2)
return all_rows
def volume_band(volume):
"""按手数分档。"""
if volume >= Decimal("6000"):
return ">6000手"
if volume >= Decimal("3000"):
return "3000-6000手"
if volume >= Decimal("1000"):
return "1000-3000手"
if volume >= Decimal("500"):
return "500-1000手"
return None
def amount_band(amount_wan):
"""按金额分档。"""
if amount_wan >= Decimal("300"):
return ">300万"
if amount_wan >= Decimal("100"):
return "100-300万"
if amount_wan >= Decimal("50"):
return "50-100万"
return None
def empty_stat():
return {
"买入金额": Decimal("0"),
"卖出金额": Decimal("0"),
"净买入": Decimal("0"),
"买入笔数": 0,
"卖出笔数": 0,
}
def add_stat(stat, band, side, amount):
if band not in stat:
stat[band] = empty_stat()
if side == "买入":
stat[band]["买入金额"] += amount
stat[band]["买入笔数"] += 1
elif side == "卖出":
stat[band]["卖出金额"] += amount
stat[band]["卖出笔数"] += 1
stat[band]["净买入"] = stat[band]["买入金额"] - stat[band]["卖出金额"]
def build_stats(records):
"""只统计新增委托。"""
volume_stat = {}
amount_stat = {}
for item in records:
if item["动作"] != "新增委托":
continue
side = item["方向"]
amount = item["金额万"]
v_band = volume_band(item["手数"])
if v_band:
add_stat(volume_stat, v_band, side, amount)
a_band = amount_band(amount)
if a_band:
add_stat(amount_stat, a_band, side, amount)
return volume_stat, amount_stat
def print_detail(records, limit=20):
print("\n========== 解析明细 ==========")
for i, item in enumerate(records[:limit], start=1):
print(
f"{i:>2}. {item['时间']} | {item['动作']} | {item['方向']} | "
f"价格={item['价格']} | 手数={item['手数']} | "
f"金额={item['金额万']:.2f}万 | 编号={item['委托编号']}"
)
def print_volume_stat(stat):
print("\n========== 成交/委托分布 ==========")
print(f"{'手数区间':<14}{'买入笔数':>10}{'卖出笔数':>10}{'净买入(万)':>14}")
for band in [">6000手", "3000-6000手", "1000-3000手", "500-1000手"]:
item = stat.get(band, empty_stat())
print(
f"{band:<14}"
f"{item['买入笔数']:>10}"
f"{item['卖出笔数']:>10}"
f"{item['净买入']:>14.2f}"
)
def print_amount_stat(stat):
print("\n========== 净买入统计 ==========")
print(f"{'金额区间':<14}{'净买入(万)':>14}{'买入金额(万)':>14}{'卖出金额(万)':>14}")
for band in [">300万", "100-300万", "50-100万"]:
item = stat.get(band, empty_stat())
print(
f"{band:<14}"
f"{item['净买入']:>14.2f}"
f"{item['买入金额']:>14.2f}"
f"{item['卖出金额']:>14.2f}"
)
def main():
raw_rows = fetch_all_rows(STOCK_ID)
records = [parse_row(row) for row in raw_rows]
print_detail(records)
new_count = sum(1 for x in records if x["动作"] == "新增委托")
cancel_count = sum(1 for x in records if x["动作"] == "撤单")
print("\n========== 摘要 ==========")
print(f"股票代码：{STOCK_ID}")
print(f"解析记录数：{len(records)}")
print(f"新增委托数：{new_count}")
print(f"撤单记录数：{cancel_count}")
volume_stat, amount_stat = build_stats(records)
print_volume_stat(volume_stat)
print_amount_stat(amount_stat)
if __name__ == "__main__":
main()
四、运行方式与结果
pip install requests
python 开盘啦委托大单_简洁版.py
五、看结果时重点看什么
- 先看新增委托和撤单数量，撤单很多时，说明盘口显示过的大单并不一定都留下来了；
- 再看 1000 手以上的大单分布，判断大单主要集中在哪个手数档；
- 最后看金额分档的净买入，判断大额买入和大额卖出哪边更强；
- 如果买卖方向和软件界面相反，只需要把 SIDE_MAP 里的 1 和 2 对调。
六、注意
这个工具只适合做数据观察和技术学习，不自动交易，也不生成买卖建议。盘中数据接口和字段可能变化，实盘使用前一定要和软件界面逐项校准。
