怎么用python程序量化板块个股领涨性
昨天发了怎么用Python量化板块辨识度个股 这篇文章，有同学问我怎么识别板块个股领涨性。
我当时提到，可以通过首次涨停时间来判断， 一般来说涨停越早的个股 直线拉板 溢价率更高， 其实这个描述直观看没什么太大问题。 但忽略了板块连板身板龙头。
怎么说呢，涨停时间 只考虑的是 板块同一起跑线， 有些个股可能提前启动涨停， 那这时候的龙一 可能已经是连板龙头了。
原本想通过 某个时间的涨幅来判断 ， 但主板、创业板规则不一致，这样考虑的规则需要很多。 既然提到龙一龙二， 那干脆直接用开盘啦的数据了，见顶部截图。因为我知道很多人玩短线看开盘啦， 用的人多了就成为大家的参考了。 但这种方式也可能遇到被量化针对爆头的情况， 毕竟盈亏同源。
其他我们也可以通过同花顺问财 比如 资金强度、个股热度等其他确定判断领涨度，毕竟领涨的 是 资金热捧的， 个股热度更高的。
伪代码提供如下, 代码里 URL和链接我脱敏成了XXX，你完全可以借助fiddler或charles自行监控 完成完整代码编写。 其实很简单。 不要问我为啥不写全了， 我怕被人举报，提供思路为主。
# -*- coding: utf-8 -*-
import streamlit as st
import requests
import datetime
import pandas as pd
import json
import akshare as ak
def get_sector_data(date, k):
# 基本URL
url1 = "https://XXX.com/w1/api/index.php"
url2 = "https://XXX.com/w1/api/index.php"
# 请求头配置（完全匹配API要求）
headers = {
"Host": "XXX.com" if k == 1 else "XXX.com",
"Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
"Connection": "keep-alive",
"Accept": "*/*",
"User-Agent": "lhb/5.17.9 (com.kaipanla.www; build:0; iOS 16.6.0) Alamofire/4.9.1",
"Accept-Language": "zh-Hans-CN;q=1.0",
"Accept-Encoding": "gzip;q=1.0, compress;q=0.5"
}
# 构建请求体参数（完全匹配API格式）
params = {
"Date": date if k == 1 else datetime.date.today().strftime("%Y-%m-%d"),
"Index": "0",
"Order": "1",
"PhoneOSNew": "2",
"Type": "1",
"VerSion": "5.17.0.9",
"ZSType": "7",
"a": "XXX",
"apiv": "w38",
"c": "XXX",
"st": "20"
}
url = url1 if k == 0 else url2
# print(f"请求URL: {url}")
# print(f"请求参数: {params}")
try:
# 发送POST请求（参数放在请求体中）
response = requests.post(
url,
headers=headers,
data=params
)
if response.status_code == 200:
data = response.json()
if "list" in data and data["list"]:
sector_list = []
for item in data["list"]:
if len(item) >= 4:
sector_list.append({
"代码": item[0],
"名称": item[1],
"强度": item[2],
"涨幅%": item[3]
})
return sector_list
else:
st.error("返回数据缺少'list'字段或为空")
else:
st.error(f"API返回错误状态码: {response.status_code}")
except Exception as e:
st.error(f"获取数据时出错：{str(e)}")
return []
def get_stock_data(sector_code, date, k):
# 基本URL
url1 = "https://XXX.com/w1/api/index.php"
url2 = "https://XXX.com/w1/api/index.php"
# 请求头配置
headers = {
"Host": "XXX.com" if k == 1 else "XXX.com",
"Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
"Connection": "keep-alive",
"Accept": "*/*",
"User-Agent": "lhb/5.17.9 (com.kaipanla.www; build:0; iOS 16.6.0) Alamofire/4.9.1",
"Accept-Language": "zh-Hans-CN;q=1.0",
"Accept-Encoding": "gzip;q=1.0, compress;q=0.5"
}
# 构建请求体参数
params = {
"PlateID": sector_code,
"Date": date if k == 1 else datetime.date.today().strftime("%Y-%m-%d"),
"Index": "0",
"Order": "1",
"PhoneOSNew": "2",
"Type": "6",
"VerSion": "5.17.0.9",
"a": "XXX",
"apiv": "XXX",
"c": "XXX",
"st": "20"
}
url = url1 if k == 0 else url2
try:
# 发送POST请求（参数放在请求体中）
response = requests.post(
url,
headers=headers,
data=params # 关键修改：使用data参数
)
if response.status_code == 200:
data = response.json()
print(data) # 调试用，可以注释掉
if "list" in data and data["list"]:
stock_list = []
for item in data["list"]:
if len(item) >= 24: # 确保有足够字段
stock_list.append({
"代码": item[0],
"名称": item[1],
"涨幅%": item[6],
"价格": item[5],
"成交额": str(int(float(item[7]) / 10000)) + '万',
'实际流通值': str(int(float(item[10]) / 100000000)) + '亿',
"板块": item[4],
"连板": item[23],
"龙头": item[24],
})
return stock_list
else:
st.error("个股返回数据缺少'list'字段或为空")
except Exception as e:
st.error(f"获取个股数据时出错：{str(e)}")
return []
def get_son_plate_info(plate_id):
"""
获取指定板块的子板块列表
参数:
plate_id (str): 板块代码
返回:
list: 子板块列表，每个子板块是一个字典，包含代码、名称、强度等信息
"""
# 构建请求URL
url = "https://XXX.com/w1/api/index.php"
params = {
"DEnd": "",
"Date": "",
"PhoneOSNew": "2",
"PlateID": plate_id, # 使用传入的板块代码
"VerSion": "5.17.0.9",
"a": "SonPlate_Info",
"apiv": "w38",
"c": "ZhiShuRanking"
}
# 请求头配置
headers = {
"Host": "XXX.com",
"Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
"Connection": "keep-alive",
"Accept": "*/*",
"User-Agent": "lhb/5.17.9 (com.kaipanla.www; build:0; iOS 16.6.0) Alamofire/4.9.1",
"Accept-Language": "zh-Hans-CN;q=1.0",
"Accept-Encoding": "gzip;q=1.0, compress;q=0.5"
}
try:
# 发送GET请求
response = requests.get(url, params=params, headers=headers)
response.raise_for_status() # 如果请求失败则抛出异常
# 解析JSON响应
data = response.json()
print(data)
# 检查响应是否成功
if data.get("errcode") == "0" and "List" in data:
# 提取子板块列表
son_plate_list = data.get("List", [])
# 解析每个子板块信息
result = []
for item in son_plate_list:
if len(item) >= 3: # 确保有足够字段
result.append({
"代码": item[0],
"名称": item[1],
"强度": item[2]
})
return result
else:
st.error(f"获取子板块失败: {data.get('msg', '未知错误')}")
return []
except requests.exceptions.RequestException as e:
st.error(f"请求错误: {e}")
return []
except json.JSONDecodeError as e:
st.error(f"JSON解析错误: {e}")
return []
def is_trading_day(date):
"""
判断指定日期是否为交易日
"""
try:
# 获取交易日历数据
trade_date_df = ak.tool_trade_date_hist_sina()
# 将日期转换为字符串格式进行比较
date_str = date.strftime("%Y-%m-%d")
# 检查日期是否在交易日历中
return date_str in trade_date_df["trade_date"].astype(str).values
except Exception as e:
st.error(f"获取交易日历时出错：{str(e)}")
return False # 出错时默认返回非交易日
def app():
st.title("精选板块")
today = datetime.date.today()
formatted_today = today.strftime("%Y-%m-%d")
date_range = [today - datetime.timedelta(days=i) for i in range(30)]
formatted_date_range = [date.strftime("%Y-%m-%d") for date in date_range]
selected_date = st.selectbox("选择日期", formatted_date_range, index=0)
# 将选择的日期字符串转换为日期对象
selected_date_obj = datetime.datetime.strptime(selected_date, "%Y-%m-%d").date()
# 判断是否为交易日
if is_trading_day(selected_date_obj):
st.success(f"{selected_date} 是交易日，正在加载数据...")
else:
st.warning(f"⚠️ {selected_date} 是非交易日，数据无法获取")
return
# 确定数据源类型（实时/历史）
k = 0 if selected_date == formatted_today else 1
# 获取板块数据
sector_data = get_sector_data(selected_date, k)
if sector_data:
df = pd.DataFrame(sector_data)
st.dataframe(df)
# 添加板块查询功能
selected_sector_code = st.selectbox("选择板块代码查看详细信息", df["代码"].tolist())
if selected_sector_code:
# 获取子板块信息
with st.expander("查看子板块"):
son_plate_data = get_son_plate_info(selected_sector_code)
if son_plate_data:
son_plate_df = pd.DataFrame(son_plate_data)
# 按强度降序排列
son_plate_df = son_plate_df.sort_values(by="强度", ascending=False)
st.dataframe(son_plate_df)
else:
st.info("该板块没有子板块或获取子板块数据失败")
# 获取个股信息
stock_data = get_stock_data(selected_sector_code, selected_date, k)
if stock_data:
stock_df = pd.DataFrame(stock_data)
st.dataframe(stock_df)
else:
st.warning("未获取到个股数据，请检查参数或网络连接")
else:
st.error("获取板块数据失败，请稍后重试")
if __name__ == "__main__":
app()如果我的分享对你有所帮助，不吝啬给个点赞关注呗。
本文参与 腾讯云自媒体同步曝光计划，分享自微信公众号。
原始发表：2025-09-13，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号