夜雨聆风龙虎榜最不缺的就是数字。
真正缺的是一份能继续计算、筛选和复盘的数据。
网页上看一只股票很方便，但如果想横向比较几十只股票，统计不同营业部的净买入，或者把结果接进自己的量化研究流程，手工复制很快就会变成体力活。
这次我把整个流程整理成了一个 Python 文件：读取同花顺龙虎榜公开页面，解析最新交易日的个股和买卖席位，再生成一份可以直接筛选的 Excel 。
代码集中在一个文件里，但要让结果真正可用，难点并不在“发起一个网络请求”。
网页看得懂，程序未必读得对
龙虎榜页面表面上是一张表，实际至少有四个容易被忽略的问题。
第一，同一只股票可能出现不止一次。
原因可能不同，统计周期也可能不同。如果只拿股票代码当唯一键，席位明细很容易串到另一条上榜记录。脚本因此给每条记录生成独立的事件 ID ，并在匹配席位时优先识别“一日”与“三日”统计周期。
第二，金额单位不统一。
网页里可能同时出现“亿”“万”和“元”。如果不先标准化，后续排序和求和就会失真。脚本把金额统一换算成“万元”，涨跌幅则转成 Excel 能正确识别的百分比数值。
第三，汇总记录和席位明细不是一一对应。
以 2026 年 8 月 7 日页面的一次测试为例，脚本解析到 67 条上榜事件，其中 46 条在首页包含席位明细，共得到 455 行席位记录。没有嵌入明细的事件不会被强行补齐，而是明确标记为“网页未提供”。
这点很重要：缺失就是缺失。数据工具最怕的不是空值，而是把空值悄悄变成一个看似完整的答案。
第四，能生成 Excel ，不等于 Excel 能稳定打开。
我在测试中遇到过一个很隐蔽的问题：同一片数据区域既设置工作表级自动筛选，又创建带筛选器的 Excel 表对象。部分 Excel 版本打开时会提示文件有问题，并删除表对象里的筛选功能。
修复方法很直接：只保留 Table 自带的筛选器，不再叠加 worksheet 的 auto_filter 。表格看起来没有变化，但文件结构不再冲突。
最终得到三张表
脚本输出的工作簿包含三个工作表。
1. 个股汇总
每一行代表一条独立的上榜事件，包括：
净额为正时标记为净买入，为负时标记为净卖出。涨跌幅和资金净额还加了条件格式，浏览时更直观。
2. 席位明细
这里保存买入前五和卖出前五营业部，包括方向、排名、营业部名称、标签、买入额、卖出额和净额。
保留“买入榜”和“卖出榜”的原始方向，而不是先替读者合并，是为了让后续研究有更多自由度。例如可以自己统计某营业部在不同榜单方向上的行为，也可以按事件 ID 回连个股汇总。
3. 说明
这张表记录数据来源、交易日、事件数量、席位行数和金额单位，并加入简单校验公式。
很多脚本只负责把文件写出来，至于写得对不对，全靠肉眼。我的习惯是至少把行数、工作表和关键公式再读一遍。脚本最后会重新打开生成的文件进行校验，发现异常就直接报错。
为什么没有直接用 pandas.read_html
对于结构规整的 HTML 表格， read_html 确实方便。
但这个页面的表头、汇总数据和席位详情并不是一张完整的标准表。席位数据还分散在多个股票详情块里，需要结合股票代码、出现顺序、统计周期和上榜原因做事件级匹配。
所以这里选择 BeautifulSoup 解析页面结构，再用 openpyxl 控制工作簿格式。代码稍长一点，换来的是更清楚的数据边界。
怎么运行
先安装依赖：
pip install requests beautifulsoup4 lxml openpyxl
把文末代码保存为 ths_longhu_fetch.py，然后运行：
python ths_longhu_fetch.py
默认会在当前目录生成：
同花顺龙虎榜_交易日期.xlsx
也可以指定 Excel 路径，并同时保存一份 JSON ：
python ths_longhu_fetch.py -o 龙虎榜.xlsx --json-output 龙虎榜.json
它适合做什么
这份数据可以作为龙虎榜研究的第一层底表，例如：
但龙虎榜只披露满足特定条件的交易信息，它不是完整资金流，更不是一个可以单独使用的买卖信号。数据整理提高的是研究效率，不会自动提高结论质量。
另外，脚本依赖网页当前结构。网站改版后， CSS 选择器和字段匹配规则可能需要调整。使用时请遵守网站规则，控制访问频率，不要用于高频、商业化或影响网站正常服务的采集。
数据来源：同花顺龙虎榜[1]。本文仅作技术交流和研究记录，不构成任何投资建议。
内容说明：本文在资料整理和文字表达环节使用人工智能辅助，数据、代码与结论边界均经人工核验。
完整代码
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import re
from collections import defaultdict, deque
from pathlib import Path
import requests
from bs4 import BeautifulSoup
from openpyxl import Workbook, load_workbook
from openpyxl.formatting.rule import CellIsRule, FormulaRule
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.worksheet.table import Table, TableStyleInfo
SOURCE_URL = "https://data.10jqka.com.cn/market/longhu/"
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0 Safari/537.36"
),
"Referer": "https://data.10jqka.com.cn/",
}
def amount_to_wan(text: str) -> float | None:
value = (
text.replace(",", "")
.replace(" ", "")
.replace("亿元", "亿")
.replace("万元", "万")
.strip()
)
if not value or value in {"--", "-"}:
return None
match = re.fullmatch(r"(-?[\d.]+)(亿|万|元)?", value)
if not match:
return None
number = float(match.group(1))
unit = match.group(2)
if unit == "亿":
return number * 10000
if unit == "元":
return number / 10000
return number
def percent_to_decimal(text: str) -> float | None:
value = text.replace("%", "").strip()
return float(value) / 100 if value and value not in {"--", "-"} else None
def number(text: str) -> float | None:
value = text.replace(",", "").strip()
return float(value) if value and value not in {"--", "-"} else None
def text_of(node) -> str:
return node.get_text(" ", strip=True) if node else ""
def parse_page(html: str) -> dict:
soup = BeautifulSoup(html, "lxml")
date_node = soup.select_one("input.startday")
trade_date = date_node.get("value", "") if date_node else ""
summary_header_table = None
for table in soup.find_all("table"):
headers = [text_of(th) for th in table.find_all("th")]
if {"代码", "名称", "现价", "涨跌幅", "成交金额", "净买入额"}.issubset(headers):
summary_header_table = table
break
if summary_header_table is None:
raise RuntimeError("未找到龙虎榜个股汇总表，页面结构可能已变化")
summary_container = summary_header_table.find_next_sibling("div")
summary_table = summary_container.find("table") if summary_container else None
if summary_table is None:
raise RuntimeError("未找到龙虎榜个股汇总数据表，页面结构可能已变化")
summary_rows = []
for tr in summary_table.select("tbody tr"):
cells = [text_of(td) for td in tr.find_all("td", recursive=False)]
if len(cells) < 7:
continue
period, code, name, price, change, turnover, net_buy = cells[:7]
summary_rows.append(
{
"period": period or "1日",
"code": code.zfill(6),
"name": name,
"price": number(price),
"change_pct": percent_to_decimal(change),
"turnover_wan": amount_to_wan(turnover),
"net_buy_wan": amount_to_wan(net_buy),
}
)
occurrence = defaultdict(int)
events = []
fallback_queues: dict[str, deque] = defaultdict(deque)
for item in summary_rows:
code = item["code"]
event_id = f"{code}_{occurrence[code]}"
occurrence[code] += 1
event = {
"trade_date": trade_date,
"event_id": event_id,
**item,
"reason": "",
"buy_total_wan": None,
"sell_total_wan": None,
"source_url": SOURCE_URL,
"has_seat_detail": False,
}
events.append(event)
fallback_queues[code].append(event)
seats = []
for div in soup.select("div.stockcont[stockcode]"):
code = div.get("stockcode", "").zfill(6)
rid = div.get("rid", "")
heading = text_of(div.find("p", recursive=False))
reason = heading.split("明细：", 1)[1] if "明细：" in heading else heading
preferred_period = "3日" if "连续三个交易日" in reason else "1日"
event = next(
(item for item in fallback_queues[code] if item["period"] == preferred_period and not item["has_seat_detail"]),
None,
)
if event is None:
event = next((item for item in fallback_queues[code] if not item["has_seat_detail"]), None)
if event is None:
continue
detail = div.select_one("div.cell-cont.cjmx")
detail_text = text_of(detail.find("p", recursive=False)) if detail else ""
extracted = {}
for label in ("成交额", "合计买入", "合计卖出", "净额"):
match = re.search(label + r"：\s*(-?[\d.]+)\s*(亿元|万元|亿|万)", detail_text)
extracted[label] = amount_to_wan(match.group(1) + match.group(2)) if match else None
event.update({
"reason": reason,
"turnover_wan": extracted["成交额"] or event.get("turnover_wan"),
"buy_total_wan": extracted["合计买入"],
"sell_total_wan": extracted["合计卖出"],
"net_buy_wan": extracted["净额"] if extracted["净额"] is not None else event.get("net_buy_wan"),
"has_seat_detail": True,
})
if not detail:
continue
for table_index, table in enumerate(detail.find_all("table", recursive=False)):
direction = "买入前五" if table_index == 0 else "卖出前五"
for rank, tr in enumerate(table.select("tbody tr"), start=1):
cells = tr.find_all("td", recursive=False)
if len(cells) < 4:
continue
broker_link = cells[0].find("a")
broker = broker_link.get("title") if broker_link and broker_link.get("title") else text_of(cells[0])
label_node = cells[0].select_one("label")
label = text_of(label_node)
if label and broker.endswith(label):
broker = broker[: -len(label)].strip()
seats.append(
{
"trade_date": trade_date,
"event_id": event["event_id"],
"code": code,
"name": event["name"],
"reason": reason,
"direction": direction,
"rank": rank,
"broker": broker,
"label": label,
"buy_wan": amount_to_wan(text_of(cells[1])),
"sell_wan": amount_to_wan(text_of(cells[2])),
"net_wan": amount_to_wan(text_of(cells[3])),
"source_url": SOURCE_URL,
}
)
return {
"metadata": {
"source_url": SOURCE_URL,
"trade_date": trade_date,
"event_count": len(events),
"detail_event_count": sum(1 for item in events if item["has_seat_detail"]),
"seat_row_count": len(seats),
},
"events": events,
"seats": seats,
}
DARK = "17365D"
PALE = "F3F6FA"
RED = "C00000"
GREEN = "008000"
WHITE = "FFFFFF"
def style_title(ws, cell_range: str, title: str) -> None:
ws.merge_cells(cell_range)
cell = ws[cell_range.split(":", 1)[0]]
cell.value = title
cell.fill = PatternFill("solid", fgColor=DARK)
cell.font = Font(color=WHITE, bold=True, size=16)
cell.alignment = Alignment(vertical="center")
ws.row_dimensions[cell.row].height = 30
def style_header(ws, row: int, start_col: int, end_col: int) -> None:
for cells in ws.iter_cols(min_col=start_col, max_col=end_col, min_row=row, max_row=row):
for cell in cells:
cell.fill = PatternFill("solid", fgColor=DARK)
cell.font = Font(color=WHITE, bold=True)
cell.alignment = Alignment(vertical="center")
ws.row_dimensions[row].height = 26
def add_table(ws, ref: str, name: str) -> None:
table = Table(displayName=name, ref=ref)
table.tableStyleInfo = TableStyleInfo(
name="TableStyleMedium2",
showFirstColumn=False,
showLastColumn=False,
showRowStripes=True,
showColumnStripes=False,
)
ws.add_table(table)
def add_source_hyperlinks(ws, column: int, first_row: int, last_row: int) -> None:
for row in range(first_row, last_row + 1):
cell = ws.cell(row=row, column=column)
if cell.value:
cell.hyperlink = str(cell.value)
cell.style = "Hyperlink"
def build_workbook(data: dict, output: Path) -> None:
wb = Workbook()
summary = wb.active
summary.title = "个股汇总"
seats = wb.create_sheet("席位明细")
notes = wb.create_sheet("说明")
meta = data["metadata"]
for ws in (summary, seats, notes):
ws.sheet_view.showGridLines = False
style_title(summary, "A1:M1", "同花顺龙虎榜数据")
summary.append([
"交易日", meta["trade_date"], "事件数", meta["event_count"],
"有席位明细事件", meta["detail_event_count"], "席位明细行数",
meta["seat_row_count"], "来源", meta["source_url"], None, None, None,
])
for cell in summary[2]:
cell.fill = PatternFill("solid", fgColor=PALE)
summary_headers = [
"统计周期", "代码", "名称", "现价", "涨跌幅", "上榜原因", "成交额(万元)",
"合计买入(万元)", "合计卖出(万元)", "净额(万元)", "资金方向", "席位明细状态", "来源URL",
]
summary.append([])
summary.append([])
summary.append(summary_headers)
for index, item in enumerate(data["events"], start=6):
summary.append([
item["period"], item["code"], item["name"], item["price"], item["change_pct"],
item["reason"], item["turnover_wan"], item["buy_total_wan"], item["sell_total_wan"],
item["net_buy_wan"], f'=IF(J{index}>0,"净买入",IF(J{index}<0,"净卖出","持平"))',
"已获取" if item["has_seat_detail"] else "网页未提供", item["source_url"],
])
summary_end = 5 + len(data["events"])
style_header(summary, 5, 1, 13)
add_table(summary, f"A5:M{summary_end}", "LonghuSummary")
summary.freeze_panes = "D6"
widths = {"A": 10, "B": 10, "C": 14, "D": 11, "E": 11, "F": 55,
"G": 16, "H": 16, "I": 16, "J": 16, "K": 11, "L": 14, "M": 32}
for column, width in widths.items():
summary.column_dimensions[column].width = width
for row in range(6, summary_end + 1):
summary.cell(row, 2).number_format = "000000"
summary.cell(row, 4).number_format = "0.00"
summary.cell(row, 5).number_format = "0.00%;[Green](0.00%);-"
for column in range(7, 11):
summary.cell(row, column).number_format = "#,##0.00;[Green](#,##0.00);-"
red_font = Font(color=RED)
green_font = Font(color=GREEN)
summary.conditional_formatting.add(f"E6:E{summary_end}", CellIsRule(operator="greaterThan", formula=["0"], font=red_font))
summary.conditional_formatting.add(f"E6:E{summary_end}", CellIsRule(operator="lessThan", formula=["0"], font=green_font))
summary.conditional_formatting.add(f"J6:J{summary_end}", CellIsRule(operator="greaterThan", formula=["0"], font=red_font))
summary.conditional_formatting.add(f"J6:J{summary_end}", CellIsRule(operator="lessThan", formula=["0"], font=green_font))
add_source_hyperlinks(summary, 13, 6, summary_end)
style_title(seats, "A1:M1", "龙虎榜买卖席位明细")
seat_headers = [
"交易日", "事件ID", "代码", "名称", "上榜原因", "榜单方向", "排名", "营业部", "标签",
"买入额(万元)", "卖出额(万元)", "净额(万元)", "来源URL",
]
seats.append([])
seats.append([])
seats.append(seat_headers)
for item in data["seats"]:
seats.append([
item["trade_date"], item["event_id"], item["code"], item["name"], item["reason"],
item["direction"], item["rank"], item["broker"], item["label"], item["buy_wan"],
item["sell_wan"], item["net_wan"], item["source_url"],
])
seats_end = 4 + len(data["seats"])
style_header(seats, 4, 1, 13)
add_table(seats, f"A4:M{seats_end}", "LonghuSeats")
seats.freeze_panes = "E5"
seat_widths = {"A": 12, "B": 15, "C": 10, "D": 14, "E": 50, "F": 11,
"G": 8, "H": 44, "I": 12, "J": 16, "K": 16, "L": 16, "M": 32}
for column, width in seat_widths.items():
seats.column_dimensions[column].width = width
for row in range(5, seats_end + 1):
seats.cell(row, 3).number_format = "000000"
for column in range(10, 13):
seats.cell(row, column).number_format = "#,##0.00;[Green](#,##0.00);-"
seats.conditional_formatting.add(f"L5:L{seats_end}", CellIsRule(operator="greaterThan", formula=["0"], font=red_font))
seats.conditional_formatting.add(f"L5:L{seats_end}", CellIsRule(operator="lessThan", formula=["0"], font=green_font))
add_source_hyperlinks(seats, 13, 5, seats_end)
style_title(notes, "A1:D1", "数据说明与校验")
notes.append([])
note_rows = [
["项目", "内容"], ["数据来源", meta["source_url"]], ["页面交易日", meta["trade_date"]],
["个股事件数", meta["event_count"]], ["有席位明细事件", meta["detail_event_count"]],
["席位明细行数", meta["seat_row_count"]], ["金额单位", "万元；原网页中的亿元已统一乘以10,000"],
["说明", "同一股票可能因不同上榜原因出现多条事件；买入榜与卖出榜中的同一席位会分别保留。首页未嵌入明细的事件会标记为“网页未提供”。"],
]
for row in note_rows:
notes.append(row)
style_header(notes, 3, 1, 2)
notes.append(["校验项", "实际值", "期望值", "状态"])
notes.append(["个股汇总行数", len(data["events"]), meta["event_count"], '=IF(B12=C12,"OK","ERROR")'])
notes.append(["席位明细行数", len(data["seats"]), meta["seat_row_count"], '=IF(B13=C13,"OK","ERROR")'])
notes.append(["个股代码长度", 6, 6, '=IF(B14=C14,"OK","ERROR")'])
notes.append(["总体状态", None, None, '=IF(COUNTIF(D12:D14,"ERROR")=0,"OK","ERROR")'])
style_header(notes, 11, 1, 4)
notes.column_dimensions["A"].width = 20
notes.column_dimensions["B"].width = 72
notes.column_dimensions["C"].width = 16
notes.column_dimensions["D"].width = 16
notes["B4"].hyperlink = meta["source_url"]
notes["B4"].style = "Hyperlink"
for row in notes.iter_rows(min_row=3, max_row=15, min_col=1, max_col=4):
for cell in row:
cell.alignment = Alignment(vertical="center", wrap_text=True)
notes.conditional_formatting.add("D12:D15", FormulaRule(formula=['D12="OK"'], fill=PatternFill("solid", fgColor="E2F0D9"), font=Font(color=GREEN, bold=True)))
notes.conditional_formatting.add("D12:D15", FormulaRule(formula=['D12="ERROR"'], fill=PatternFill("solid", fgColor="FCE4D6"), font=Font(color=RED, bold=True)))
output.parent.mkdir(parents=True, exist_ok=True)
wb.save(output)
def validate_workbook(path: Path, data: dict) -> None:
wb = load_workbook(path, data_only=False, read_only=True)
required = {"个股汇总", "席位明细", "说明"}
if set(wb.sheetnames) != required:
raise RuntimeError(f"工作表校验失败：{wb.sheetnames}")
if wb["个股汇总"].max_row - 5 != data["metadata"]["event_count"]:
raise RuntimeError("个股汇总行数校验失败")
if wb["席位明细"].max_row - 4 != data["metadata"]["seat_row_count"]:
raise RuntimeError("席位明细行数校验失败")
if not str(wb["个股汇总"]["K6"].value).startswith("=IF("):
raise RuntimeError("资金方向公式校验失败")
wb.close()
def main() -> None:
parser = argparse.ArgumentParser(description="获取同花顺龙虎榜数据并生成 Excel")
parser.add_argument("-o", "--output", help="Excel 输出路径；默认按交易日命名")
parser.add_argument("--json-output", help="可选：同时保存原始结构化 JSON")
args = parser.parse_args()
response = requests.get(SOURCE_URL, headers=HEADERS, timeout=30)
response.raise_for_status()
response.encoding = "gbk"
result = parse_page(response.text)
output = Path(args.output) if args.output else Path(f"同花顺龙虎榜_{result['metadata']['trade_date']}.xlsx")
if output.suffix.lower() != ".xlsx":
output = output.with_suffix(".xlsx")
build_workbook(result, output)
validate_workbook(output, result)
if args.json_output:
json_output = Path(args.json_output)
json_output.parent.mkdir(parents=True, exist_ok=True)
json_output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps({**result["metadata"], "excel": str(output.resolve())}, ensure_ascii=False))
if __name__ == "__main__":
main()