394 lines
18 KiB
Python
394 lines
18 KiB
Python
# 导入pandas库，用于数据处理和分析，简称pd
|
|
import pandas as pd
|
|
# 导入requests库，用于发送HTTP网络请求
|
|
import requests
|
|
# 导入json库，用于处理JSON格式的数据
|
|
import json
|
|
# 导入warnings库，用于处理警告信息
|
|
import warnings
|
|
# 导入yaml库，用于读取配置文件
|
|
import yaml
|
|
# 导入os库，用于文件路径操作
|
|
import os
|
|
# 忽略所有警告信息，让程序运行更清爽
|
|
warnings.filterwarnings(action='ignore')
|
|
# 定义雪球数据获取类
|
|
class xueqie_data:
|
|
# 类的初始化方法，创建对象时自动调用
|
|
def __init__(self, cookie_list=None, config=None):
|
|
'''
|
|
雪球数据获取类的初始化方法
|
|
参数说明:
|
|
cookie_list: 雪球网站的登录凭证列表，用于模拟用户登录
|
|
config: 配置信息字典
|
|
'''
|
|
print('[初始化] 正在初始化雪球数据获取工具...')
|
|
|
|
# 保存配置信息
|
|
self.config = config if config else load_config()
|
|
|
|
# 从配置文件或参数获取cookie列表
|
|
if cookie_list is not None:
|
|
self.cookie_list = cookie_list
|
|
print('[参数] 使用传入的cookie列表')
|
|
else:
|
|
# 从配置文件读取cookie列表
|
|
self.cookie_list = self.config.get('cookie列表', [])
|
|
print('[配置] 从配置文件读取cookie列表')
|
|
|
|
print('[加载] 已加载{}个身份验证信息'.format(len(self.cookie_list)))
|
|
|
|
# 创建一个空的DataFrame用于存储获取到的数据
|
|
self.df=pd.DataFrame()
|
|
print('[完成] 雪球数据获取工具初始化完成')
|
|
# 定义获取HTTP请求头的方法
|
|
def get_headers(self,cookie=''):
|
|
'''
|
|
构造访问雪球网站时需要的HTTP请求头信息
|
|
参数说明:
|
|
cookie: 用户的登录凭证字符串
|
|
返回值: 包含所有必要头信息的字典
|
|
'''
|
|
# 构造HTTP请求头字典，模拟真实浏览器访问
|
|
headers={
|
|
# 告诉服务器可以接受任何类型的响应内容
|
|
'Accept':'*/*',
|
|
# 告诉服务器支持的压缩格式，用于减少传输数据量
|
|
'Accept-Encoding':'gzip, deflate, br, zstd',
|
|
# 设置语言偏好：首选中文，其次英文
|
|
'Accept-Language':'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
|
|
# 保持HTTP连接，提高效率
|
|
'Connection':'keep-alive',
|
|
# 用户的登录凭证，最重要的认证信息
|
|
'Cookie':cookie,
|
|
# 目标网站的域名
|
|
'Host':'xueqiu.com',
|
|
# 告诉服务器请求来源页面，防止被识别为爬虫
|
|
'Referer':'https://xueqiu.com/P/ZH3223683',
|
|
# 浏览器安全相关的头信息，模拟Chrome/Edge浏览器
|
|
'Sec-Ch-Ua':'"Chromium";v="124", "Microsoft Edge";v="124", "Not-A.Brand";v="99"',
|
|
# 表示不是移动设备
|
|
'Sec-Ch-Ua-Mobile':'?0',
|
|
# 操作系统平台信息
|
|
'Sec-Ch-Ua-Platform':"Windows",
|
|
# 请求目标类型
|
|
'Sec-Fetch-Dest':'empty',
|
|
# 请求模式，CORS表示跨域请求
|
|
'Sec-Fetch-Mode':'cors',
|
|
# 请求来源，same-origin表示同源请求
|
|
'Sec-Fetch-Site':'same-origin',
|
|
# 用户代理字符串，模拟Edge浏览器，让服务器认为是真实用户
|
|
'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0',
|
|
# 表示这是一个AJAX请求
|
|
'X-Requested-With':'XMLHttpRequest'
|
|
}
|
|
# 返回构造好的请求头字典
|
|
return headers
|
|
# 定义获取最近调仓记录的方法
|
|
def get_the_latest_move(self,rb_id=''):
|
|
'''
|
|
获取投资组合的最近一次调仓记录
|
|
参数说明:
|
|
rb_id: 调仓记录的ID，用于获取特定的调仓信息
|
|
返回值: 包含调仓记录的DataFrame，如果失败则返回空DataFrame
|
|
API地址示例: https://xueqiu.com/cubes/rebalancing/show_origin.json?rb_id=164306198&cube_symbol=ZH3223683
|
|
'''
|
|
# 雪球API的URL地址，用于获取调仓详情
|
|
url='https://xueqiu.com/cubes/rebalancing/show_origin.json?'
|
|
# 获取HTTP请求头，但这里没有传入cookie（可能是个bug）
|
|
headers=self.get_headers()
|
|
# 构造请求参数
|
|
params={
|
|
'rb_id':rb_id, # 调仓记录ID
|
|
'cube_symbol':self.assembly_id # 投资组合ID（注意：这里使用了未定义的属性）
|
|
}
|
|
# 使用try-except处理可能的网络错误
|
|
try:
|
|
# 发送GET请求获取数据
|
|
print('[获取] 正在获取最新调仓信息...')
|
|
res=requests.get(url=url,headers=headers,params=params)
|
|
# 将响应转换为JSON格式
|
|
text=res.json()
|
|
|
|
# 检查是否有错误码
|
|
if 'error_code' in text:
|
|
print('[错误] 获取最新调仓失败: {}'.format(text.get('error_description', '未知错误')))
|
|
return pd.DataFrame()
|
|
|
|
# 检查API返回的状态
|
|
stats=text['rebalancing']['status']
|
|
# 如果请求成功
|
|
if stats=='success':
|
|
print('[成功] 最新调仓数据获取成功')
|
|
# 提取调仓历史记录
|
|
result=text['rebalancing']['rebalancing_histories']
|
|
# 转换为DataFrame格式
|
|
df=pd.DataFrame(result)
|
|
# <<< 关键：过滤未成交/已取消的单子 >>>
|
|
print(f'[过滤] 已过滤掉 {len(result) - len(df)} 条未成交/已取消的调仓记录')
|
|
df = df[df['price'].notna()]
|
|
# 将时间戳转换为可读的日期时间格式（毫秒级时间戳）
|
|
df['updated_at']=pd.to_datetime(df['updated_at'],unit='ms')
|
|
print('[数据] 成功获取{}条最新调仓记录'.format(len(df)))
|
|
return df
|
|
else:
|
|
# 如果请求失败，打印错误信息
|
|
print('[失败] 最新调仓数据获取失败')
|
|
print('[响应] 服务器响应: {}'.format(text))
|
|
print('[建议] 建议检查身份验证信息或稍后重试')
|
|
# 返回空的DataFrame
|
|
df=pd.DataFrame()
|
|
return df
|
|
# 捕获所有异常
|
|
except Exception as e:
|
|
# 打印异常信息
|
|
print('[异常] 获取最新调仓数据时发生错误: {}'.format(str(e)))
|
|
print('[建议] 建议检查网络连接或身份验证信息')
|
|
# 返回空的DataFrame
|
|
df=pd.DataFrame()
|
|
return df
|
|
# 定义获取历史调仓记录的方法
|
|
def get_hist_move(self,assembly_id=None):
|
|
'''
|
|
获取投资组合的历史调仓记录
|
|
参数说明:
|
|
assembly_id: 投资组合ID，如果为None则从配置文件读取
|
|
返回值: 包含所有历史调仓记录的DataFrame
|
|
API地址示例: https://xueqiu.com/cubes/rebalancing/history.json?cube_symbol=ZH3223683&count=20&page=1
|
|
'''
|
|
# 如果没有传入组合ID，从组合配置中读取
|
|
if assembly_id is None:
|
|
portfolio_configs = self.config.get('组合配置', [])
|
|
if portfolio_configs:
|
|
assembly_id = portfolio_configs[0].get('组合ID', 'ZH3361149')
|
|
else:
|
|
assembly_id = 'ZH3361149' # 修正为实际数据中的组合ID
|
|
print(f'[配置] 从配置文件读取组合ID: {assembly_id}')
|
|
# 初始化cookie计数器，用于跟踪尝试了第几个cookie
|
|
j=1
|
|
# 状态标志，True表示还需要继续尝试，False表示已经成功获取数据
|
|
stats=True
|
|
# 遍历所有可用的cookie，尝试获取数据
|
|
for cookie in self.cookie_list:
|
|
# 如果还没有成功获取数据，继续尝试
|
|
if stats:
|
|
# 雪球历史调仓API的URL地址
|
|
url='https://xueqiu.com/cubes/rebalancing/history.json?'
|
|
# 使用当前cookie构造请求头
|
|
headers=self.get_headers(cookie=cookie)
|
|
# 构造请求参数
|
|
params={
|
|
'cube_symbol':assembly_id, # 投资组合ID
|
|
'count': '50', # 每页返回50条记录
|
|
'page': '1', # 获取第1页数据
|
|
}
|
|
# 发送GET请求
|
|
res=requests.get(url=url,headers=headers,params=params)
|
|
# 将响应转换为JSON格式
|
|
text=res.json()
|
|
|
|
# 获取响应中的所有键名，用于检查是否有错误
|
|
keys_list=list(set(text.keys()))
|
|
# 如果响应中没有'error_code'键，说明请求成功
|
|
if 'error_code' not in keys_list:
|
|
# 保存原始数据到JSON文件，便于分析数据结构
|
|
import json
|
|
with open('原始数据.json', 'w', encoding='utf-8') as f:
|
|
json.dump(text, f, indent=2, ensure_ascii=False)
|
|
print('[保存] 雪球原始数据已保存至: 原始数据.json')
|
|
|
|
# 将调仓列表转换为DataFrame
|
|
df=pd.DataFrame(text['list'])
|
|
# 创建空的DataFrame用于存储所有调仓记录
|
|
data=pd.DataFrame()
|
|
# 遍历每个调仓操作及其历史记录
|
|
for idx, rebalancing in df.iterrows():
|
|
histories = rebalancing['rebalancing_histories']
|
|
|
|
# 将每个调仓历史转换为DataFrame
|
|
df1 = pd.DataFrame(histories)
|
|
print(f'[过滤] 已过滤掉 {len(histories) - len(df1)} 条未成交/已取消的调仓记录')
|
|
# <<< 关键修改：过滤已取消或未成交的单子 >>>
|
|
df1 = df1[df1['price'].notna()] # 过滤掉 price 为 None 的已取消/未成交单
|
|
|
|
# 添加组合ID和组合名字信息
|
|
if len(df1) > 0:
|
|
df1['组合ID'] = assembly_id
|
|
# 获取组合名字
|
|
portfolio_name = self.get_portfolio_name(assembly_id)
|
|
df1['组合名字'] = portfolio_name
|
|
|
|
# 将所有调仓记录合并到一个DataFrame中
|
|
data=pd.concat([data,df1],ignore_index=True)
|
|
print('[成功] 第{}个身份验证成功，开始获取数据...'.format(j))
|
|
# 设置状态为False，表示已经成功获取数据，不需要再尝试其他cookie
|
|
stats=False
|
|
else:
|
|
# 如果有错误，打印错误信息
|
|
print('[失败] 第{}个身份验证失败: {}'.format(j, text.get('error_description', '未知错误')))
|
|
print('[重试] 正在尝试下一个身份验证...')
|
|
else:
|
|
# 如果前面已经成功获取数据，跳过后续cookie
|
|
print('[跳过] 第{}个身份验证跳过（已获取到数据）'.format(j))
|
|
stats=False
|
|
# cookie计数器加1
|
|
j+=1
|
|
# 检查最终状态
|
|
if stats:
|
|
# 如果所有cookie都失败了，这里什么都不做
|
|
pass
|
|
else:
|
|
# 这个打印信息有误，应该是"cookie 有可用的"
|
|
print('[完成] 数据获取成功，已找到可用的身份验证')
|
|
# 使用try-except处理可能的变量未定义错误
|
|
try:
|
|
# 检查data变量是否存在且有数据
|
|
if data.shape[0]>=0:
|
|
data=data
|
|
else:
|
|
# 如果没有数据，返回空DataFrame
|
|
data=pd.DataFrame()
|
|
except:
|
|
# 如果data变量未定义，创建空DataFrame
|
|
data=pd.DataFrame()
|
|
# 返回获取到的历史调仓数据
|
|
return data
|
|
|
|
def get_portfolio_name(self, assembly_id):
|
|
"""
|
|
根据组合ID获取组合名字
|
|
"""
|
|
try:
|
|
if self.config:
|
|
portfolio_configs = self.config.get('组合配置', [])
|
|
for config in portfolio_configs:
|
|
if config.get('组合ID') == assembly_id:
|
|
return config.get('组合名字', assembly_id)
|
|
return assembly_id # 如果找不到配置，返回组合ID
|
|
except Exception as e:
|
|
print(f'[警告] 获取组合名字失败: {e}')
|
|
return assembly_id
|
|
|
|
# 定义列名翻译映射字典
|
|
def get_column_translation():
|
|
'''
|
|
返回英文列名到中文列名的映射字典
|
|
'''
|
|
column_mapping = {
|
|
'id': '记录ID',
|
|
'rebalancing_id': '调仓ID',
|
|
'组合ID': '组合ID',
|
|
'stock_id': '股票ID',
|
|
'stock_name': '股票名称',
|
|
'stock_symbol': '股票代码',
|
|
'volume': '持仓数量',
|
|
'price': '价格',
|
|
'net_value': '净值',
|
|
'weight': '当前权重',
|
|
'target_weight': '目标权重',
|
|
'prev_weight': '前期权重',
|
|
'prev_target_weight': '前期目标权重',
|
|
'prev_weight_adjusted': '前期调整权重',
|
|
'prev_volume': '前期数量',
|
|
'prev_price': '前期价格',
|
|
'prev_net_value': '前期净值',
|
|
'proactive': '主动调仓',
|
|
'created_at': '创建时间',
|
|
'updated_at': '更新时间',
|
|
'target_volume': '目标数量',
|
|
'prev_target_volume': '前期目标数量'
|
|
}
|
|
return column_mapping
|
|
|
|
# 定义配置文件读取函数
|
|
def load_config():
|
|
'''
|
|
从YAML配置文件读取参数
|
|
返回配置字典
|
|
'''
|
|
config_file = '参数设置.yaml'
|
|
try:
|
|
if os.path.exists(config_file):
|
|
with open(config_file, 'r', encoding='utf-8') as f:
|
|
config = yaml.safe_load(f)
|
|
print(f'[配置] 成功加载配置文件: {config_file}')
|
|
return config
|
|
else:
|
|
print(f'[警告] 配置文件不存在: {config_file}')
|
|
print('[默认] 使用默认配置')
|
|
return {'组合ID': 'ZH3361149'}
|
|
except Exception as e:
|
|
print(f'[错误] 读取配置文件失败: {str(e)}')
|
|
print('[默认] 使用默认配置')
|
|
return {'组合ID': 'ZH3361149'}
|
|
|
|
|
|
|
|
# 主程序入口，当直接运行此文件时执行
|
|
if __name__=='__main__':
|
|
'''
|
|
雪球投资组合数据获取程序
|
|
'''
|
|
print('[启动] 雪球跟单数据获取程序启动...')
|
|
print('[初始化] 正在初始化数据获取工具...')
|
|
|
|
# 从配置文件获取组合配置
|
|
config = load_config()
|
|
|
|
# 获取组合ID列表
|
|
portfolio_configs = config.get('组合配置', [])
|
|
if portfolio_configs:
|
|
assembly_ids = [config.get('组合ID') for config in portfolio_configs if config.get('组合ID')]
|
|
else:
|
|
assembly_ids = ['ZH3361149'] # 默认组合ID（修正为实际数据中的组合ID）
|
|
print(f'[配置] 使用组合ID: {assembly_ids}')
|
|
|
|
# 创建雪球数据获取对象
|
|
trader = xueqie_data(config=config)
|
|
|
|
# 合并所有组合的数据
|
|
all_data = pd.DataFrame()
|
|
|
|
for assembly_id in assembly_ids:
|
|
print(f'[处理] 正在获取组合 {assembly_id} 的数据...')
|
|
df = trader.get_hist_move(assembly_id=assembly_id)
|
|
if len(df) > 0:
|
|
all_data = pd.concat([all_data, df], ignore_index=True)
|
|
print(f'[成功] 组合 {assembly_id} 获取到 {len(df)} 条记录')
|
|
else:
|
|
print(f'[警告] 组合 {assembly_id} 未获取到数据')
|
|
|
|
df = all_data
|
|
# 打印当前进度，让用户了解数据获取状态
|
|
if len(df) > 0:
|
|
print('[成功] 数据获取成功！共获取{}条调仓记录'.format(len(df)))
|
|
print('[保存] 正在保存数据到CSV文件...')
|
|
|
|
# 获取列名翻译映射
|
|
column_mapping = get_column_translation()
|
|
|
|
# 翻译列名为中文
|
|
df_chinese = df.copy()
|
|
df_chinese.columns = [column_mapping.get(col, col) for col in df.columns]
|
|
|
|
# 保存为临时CSV文件，避免覆盖交易系统处理后的数据
|
|
df_chinese.to_csv('./辅助文件/雪球原始数据.csv', index=False, encoding='utf-8-sig')
|
|
print('[完成] 雪球原始数据已保存到：雪球原始数据.csv（中文列名）')
|
|
print('[提示] 交易系统将读取此文件并处理后保存到数据.csv')
|
|
|
|
# # 同时保存原始英文列名版本供参考
|
|
# df.to_csv('数据_英文列名.csv', index=False, encoding='utf-8-sig')
|
|
# print('[完成] 原始数据已保存到：数据_英文列名.csv（英文列名）')
|
|
|
|
else:
|
|
print('[警告] 本次未获取到数据，可能是网络问题或身份验证失效')
|
|
|
|
print('\n[完成] 所有数据获取任务完成！')
|
|
print('[文件] 请查看工作目录下的 CSV 文件')
|
|
print(' - 雪球原始数据.csv（雪球原始数据，中文列名版本）')
|
|
print(' - 数据.csv（交易系统处理后的完整数据）')
|
|
|
|
|
|
|
|
|
|
|