oficcejo
336 lines
11 KiB
Python
336 lines
11 KiB
Python
"""
|
|
智瞰龙虎数据采集模块
|
|
使用StockAPI获取龙虎榜数据
|
|
"""
|
|
|
|
import requests
|
|
import pandas as pd
|
|
from datetime import datetime, timedelta
|
|
import time
|
|
import warnings
|
|
|
|
warnings.filterwarnings('ignore')
|
|
|
|
|
|
class LonghubangDataFetcher:
|
|
"""龙虎榜数据获取类"""
|
|
|
|
def __init__(self, api_key=None):
|
|
"""
|
|
初始化数据获取器
|
|
|
|
Args:
|
|
api_key: StockAPI的API密钥（可选，普通请求每日免费1000次）
|
|
"""
|
|
print("[智瞰龙虎] 龙虎榜数据获取器初始化...")
|
|
# self.base_url = "https://api-lhb.zhongdu.net"
|
|
self.base_url = "http://lhb-api.ws4.cn/v1"
|
|
# self.base_url = "https://www.stockapi.com.cn/v1"
|
|
self.api_key = api_key
|
|
self.max_retries = 3 # 最大重试次数
|
|
self.retry_delay = 2 # 重试延迟（秒）
|
|
self.request_delay = 0.025 # 请求间隔（秒），40次/秒 = 0.025秒/次
|
|
|
|
def _safe_request(self, url, params=None):
|
|
"""
|
|
安全的HTTP请求，包含重试机制
|
|
|
|
Args:
|
|
url: 请求URL
|
|
params: 请求参数
|
|
|
|
Returns:
|
|
dict: 响应数据
|
|
"""
|
|
for attempt in range(self.max_retries):
|
|
try:
|
|
response = requests.get(url, params=params, timeout=10)
|
|
|
|
# 添加请求延迟，遵守40次/秒的限制
|
|
time.sleep(self.request_delay)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 20000:
|
|
return data
|
|
else:
|
|
print(f" API返回错误: {data.get('msg', '未知错误')}")
|
|
return None
|
|
else:
|
|
print(f" HTTP错误: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
if attempt < self.max_retries - 1:
|
|
print(f" 请求失败，{self.retry_delay}秒后重试... (尝试 {attempt + 1}/{self.max_retries})")
|
|
time.sleep(self.retry_delay)
|
|
else:
|
|
print(f" 请求失败，已达最大重试次数: {e}")
|
|
return None
|
|
|
|
return None
|
|
|
|
def get_longhubang_data(self, date):
|
|
"""
|
|
获取指定日期的龙虎榜数据
|
|
|
|
Args:
|
|
date: 日期，格式为 YYYY-MM-DD，如 "2023-03-21"
|
|
|
|
Returns:
|
|
dict: 龙虎榜数据
|
|
"""
|
|
print(f"[智瞰龙虎] 获取 {date} 的龙虎榜数据...")
|
|
|
|
# url = f"{self.base_url}"
|
|
url = f"{self.base_url}/youzi/all"
|
|
params = {'date': date}
|
|
|
|
result = self._safe_request(url, params)
|
|
|
|
if result and result.get('data'):
|
|
print(f" ✓ 成功获取 {len(result['data'])} 条龙虎榜记录")
|
|
return result
|
|
else:
|
|
print(f" ✗ 未获取到数据")
|
|
return None
|
|
|
|
def get_longhubang_data_range(self, start_date, end_date):
|
|
"""
|
|
获取日期范围内的龙虎榜数据
|
|
|
|
Args:
|
|
start_date: 开始日期，格式为 YYYY-MM-DD
|
|
end_date: 结束日期，格式为 YYYY-MM-DD
|
|
|
|
Returns:
|
|
list: 龙虎榜数据列表
|
|
"""
|
|
print(f"[智瞰龙虎] 获取 {start_date} 至 {end_date} 的龙虎榜数据...")
|
|
|
|
all_data = []
|
|
|
|
# 转换日期
|
|
current_date = datetime.strptime(start_date, '%Y-%m-%d')
|
|
end_date_obj = datetime.strptime(end_date, '%Y-%m-%d')
|
|
|
|
while current_date <= end_date_obj:
|
|
date_str = current_date.strftime('%Y-%m-%d')
|
|
|
|
# 跳过周末
|
|
if current_date.weekday() < 5: # 0-4表示周一到周五
|
|
result = self.get_longhubang_data(date_str)
|
|
if result and result.get('data'):
|
|
all_data.extend(result['data'])
|
|
|
|
# 下一天
|
|
current_date += timedelta(days=1)
|
|
|
|
print(f"[智瞰龙虎] ✓ 共获取 {len(all_data)} 条记录")
|
|
return all_data
|
|
|
|
def get_recent_days_data(self, days=5):
|
|
"""
|
|
获取最近N个交易日的龙虎榜数据
|
|
|
|
Args:
|
|
days: 天数（默认5天）
|
|
|
|
Returns:
|
|
list: 龙虎榜数据列表
|
|
"""
|
|
end_date = datetime.now()
|
|
start_date = end_date - timedelta(days=days * 2) # 乘以2以确保包含足够的交易日
|
|
|
|
return self.get_longhubang_data_range(
|
|
start_date.strftime('%Y-%m-%d'),
|
|
end_date.strftime('%Y-%m-%d')
|
|
)
|
|
|
|
def parse_to_dataframe(self, data_list):
|
|
"""
|
|
将龙虎榜数据转换为DataFrame
|
|
|
|
Args:
|
|
data_list: 龙虎榜数据列表
|
|
|
|
Returns:
|
|
pd.DataFrame: 数据框
|
|
"""
|
|
if not data_list:
|
|
return pd.DataFrame()
|
|
|
|
df = pd.DataFrame(data_list)
|
|
|
|
# 重命名列
|
|
column_mapping = {
|
|
'yzmc': '游资名称',
|
|
'yyb': '营业部',
|
|
'sblx': '榜单类型',
|
|
'gpdm': '股票代码',
|
|
'gpmc': '股票名称',
|
|
'mrje': '买入金额',
|
|
'mcje': '卖出金额',
|
|
'jlrje': '净流入金额',
|
|
'rq': '日期',
|
|
'gl': '概念'
|
|
}
|
|
|
|
df = df.rename(columns=column_mapping)
|
|
|
|
# 转换数据类型
|
|
numeric_columns = ['买入金额', '卖出金额', '净流入金额']
|
|
for col in numeric_columns:
|
|
if col in df.columns:
|
|
df[col] = pd.to_numeric(df[col], errors='coerce')
|
|
|
|
# 排序
|
|
if '净流入金额' in df.columns:
|
|
df = df.sort_values('净流入金额', ascending=False)
|
|
|
|
return df
|
|
|
|
def analyze_data_summary(self, data_list):
|
|
"""
|
|
分析龙虎榜数据，生成摘要统计
|
|
|
|
Args:
|
|
data_list: 龙虎榜数据列表
|
|
|
|
Returns:
|
|
dict: 统计摘要
|
|
"""
|
|
if not data_list:
|
|
return {}
|
|
|
|
df = self.parse_to_dataframe(data_list)
|
|
|
|
summary = {
|
|
'total_records': len(df),
|
|
'total_stocks': df['股票代码'].nunique() if '股票代码' in df.columns else 0,
|
|
'total_youzi': df['游资名称'].nunique() if '游资名称' in df.columns else 0,
|
|
'total_buy_amount': df['买入金额'].sum() if '买入金额' in df.columns else 0,
|
|
'total_sell_amount': df['卖出金额'].sum() if '卖出金额' in df.columns else 0,
|
|
'total_net_inflow': df['净流入金额'].sum() if '净流入金额' in df.columns else 0,
|
|
}
|
|
|
|
# Top游资排名
|
|
if '游资名称' in df.columns and '净流入金额' in df.columns:
|
|
top_youzi = df.groupby('游资名称')['净流入金额'].sum().sort_values(ascending=False)
|
|
summary['top_youzi'] = top_youzi.head(10).to_dict()
|
|
|
|
# Top股票排名
|
|
if '股票代码' in df.columns and '净流入金额' in df.columns:
|
|
top_stocks = df.groupby(['股票代码', '股票名称'])['净流入金额'].sum().sort_values(ascending=False)
|
|
summary['top_stocks'] = [
|
|
{'code': code, 'name': name, 'net_inflow': amount}
|
|
for (code, name), amount in top_stocks.head(20).items()
|
|
]
|
|
|
|
# 热门概念统计
|
|
if '概念' in df.columns:
|
|
all_concepts = []
|
|
for concepts in df['概念'].dropna():
|
|
all_concepts.extend([c.strip() for c in str(concepts).split(',')])
|
|
|
|
from collections import Counter
|
|
concept_counter = Counter(all_concepts)
|
|
summary['hot_concepts'] = dict(concept_counter.most_common(20))
|
|
|
|
return summary
|
|
|
|
def format_data_for_ai(self, data_list, summary=None):
|
|
"""
|
|
将龙虎榜数据格式化为适合AI分析的文本格式
|
|
|
|
Args:
|
|
data_list: 龙虎榜数据列表
|
|
summary: 统计摘要（可选）
|
|
|
|
Returns:
|
|
str: 格式化的文本
|
|
"""
|
|
if not data_list:
|
|
return "暂无龙虎榜数据"
|
|
|
|
df = self.parse_to_dataframe(data_list)
|
|
|
|
if summary is None:
|
|
summary = self.analyze_data_summary(data_list)
|
|
|
|
text_parts = []
|
|
|
|
# 总体概况
|
|
text_parts.append(f"""
|
|
【龙虎榜总体概况】
|
|
数据时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
|
记录总数: {summary.get('total_records', 0)}
|
|
涉及股票: {summary.get('total_stocks', 0)} 只
|
|
涉及游资: {summary.get('total_youzi', 0)} 个
|
|
总买入金额: {summary.get('total_buy_amount', 0):,.2f} 元
|
|
总卖出金额: {summary.get('total_sell_amount', 0):,.2f} 元
|
|
净流入金额: {summary.get('total_net_inflow', 0):,.2f} 元
|
|
""")
|
|
|
|
# Top游资
|
|
if summary.get('top_youzi'):
|
|
text_parts.append("\n【活跃游资 TOP10】")
|
|
for idx, (name, amount) in enumerate(summary['top_youzi'].items(), 1):
|
|
text_parts.append(f"{idx}. {name}: {amount:,.2f} 元")
|
|
|
|
# Top股票
|
|
if summary.get('top_stocks'):
|
|
text_parts.append("\n【资金净流入 TOP20股票】")
|
|
for idx, stock in enumerate(summary['top_stocks'], 1):
|
|
text_parts.append(
|
|
f"{idx}. {stock['name']}({stock['code']}): {stock['net_inflow']:,.2f} 元"
|
|
)
|
|
|
|
# 热门概念
|
|
if summary.get('hot_concepts'):
|
|
text_parts.append("\n【热门概念 TOP20】")
|
|
for idx, (concept, count) in enumerate(list(summary['hot_concepts'].items())[:20], 1):
|
|
text_parts.append(f"{idx}. {concept}: {count} 次")
|
|
|
|
# 详细交易记录（前50条）
|
|
text_parts.append("\n【详细交易记录 TOP50】")
|
|
for idx, row in df.head(50).iterrows():
|
|
text_parts.append(
|
|
f"{row.get('游资名称', 'N/A')} | "
|
|
f"{row.get('股票名称', 'N/A')}({row.get('股票代码', 'N/A')}) | "
|
|
f"买入:{row.get('买入金额', 0):,.0f} "
|
|
f"卖出:{row.get('卖出金额', 0):,.0f} "
|
|
f"净流入:{row.get('净流入金额', 0):,.0f} | "
|
|
f"日期:{row.get('日期', 'N/A')}"
|
|
)
|
|
|
|
return "\n".join(text_parts)
|
|
|
|
|
|
# 测试函数
|
|
if __name__ == "__main__":
|
|
print("=" * 60)
|
|
print("测试智瞰龙虎数据采集模块")
|
|
print("=" * 60)
|
|
|
|
fetcher = LonghubangDataFetcher()
|
|
|
|
# 测试获取单日数据
|
|
date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
|
result = fetcher.get_longhubang_data(date)
|
|
|
|
if result and result.get('data'):
|
|
# 分析数据
|
|
summary = fetcher.analyze_data_summary(result['data'])
|
|
|
|
print("\n" + "=" * 60)
|
|
print("数据采集成功！")
|
|
print("=" * 60)
|
|
|
|
# 格式化输出
|
|
formatted_text = fetcher.format_data_for_ai(result['data'], summary)
|
|
print(formatted_text[:2000]) # 显示前2000字符
|
|
print(f"\n... (总长度: {len(formatted_text)} 字符)")
|
|
else:
|
|
print("\n数据采集失败")
|
|
|