songzhuoyuan
342 lines
13 KiB
Python
342 lines
13 KiB
Python
"""
|
|
资金流向数据获取模块（akshare版本）
|
|
使用akshare的stock_individual_fund_flow接口获取个股资金流向
|
|
"""
|
|
|
|
import pandas as pd
|
|
import sys
|
|
import io
|
|
import warnings
|
|
from datetime import datetime, timedelta
|
|
import akshare as ak
|
|
from data_source_manager import data_source_manager
|
|
|
|
warnings.filterwarnings('ignore')
|
|
|
|
# 设置标准输出编码为UTF-8（仅在命令行环境，避免streamlit冲突）
|
|
def _setup_stdout_encoding():
|
|
"""仅在命令行环境设置标准输出编码"""
|
|
if sys.platform == 'win32' and not hasattr(sys.stdout, '_original_stream'):
|
|
try:
|
|
# 检测是否在streamlit环境中
|
|
import streamlit
|
|
# 在streamlit中不修改stdout
|
|
return
|
|
except ImportError:
|
|
# 不在streamlit环境，可以安全修改
|
|
try:
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='ignore')
|
|
except:
|
|
pass
|
|
|
|
_setup_stdout_encoding()
|
|
|
|
|
|
class FundFlowAkshareDataFetcher:
|
|
"""资金流向数据获取类（使用akshare数据源）"""
|
|
|
|
def __init__(self):
|
|
self.days = 30 # 获取最近30个交易日
|
|
self.available = True
|
|
print("[OK] 资金流向数据获取器初始化成功（akshare数据源）")
|
|
|
|
def get_fund_flow_data(self, symbol):
|
|
"""
|
|
获取个股资金流向数据
|
|
|
|
Args:
|
|
symbol: 股票代码（6位数字）
|
|
|
|
Returns:
|
|
dict: 包含资金流向数据的字典
|
|
"""
|
|
data = {
|
|
"symbol": symbol,
|
|
"fund_flow_data": None,
|
|
"data_success": False,
|
|
"source": "akshare"
|
|
}
|
|
|
|
# 只支持中国股票
|
|
if not self._is_chinese_stock(symbol):
|
|
data["error"] = "资金流向数据仅支持中国A股股票"
|
|
return data
|
|
|
|
try:
|
|
print(f"[资金流向] 正在获取 {symbol} 的资金流向数据...")
|
|
|
|
# 确定市场
|
|
market = self._get_market(symbol)
|
|
|
|
# 获取资金流向数据
|
|
fund_flow_data = self._get_individual_fund_flow(symbol, market)
|
|
|
|
if fund_flow_data:
|
|
data["fund_flow_data"] = fund_flow_data
|
|
print(f" [OK] 成功获取 {len(fund_flow_data.get('data', []))} 个交易日的资金流向数据")
|
|
data["data_success"] = True
|
|
print("[完成] 资金流向数据获取完成")
|
|
else:
|
|
print("[警告] 未能获取到资金流向数据")
|
|
|
|
except Exception as e:
|
|
print(f"[ERROR] 获取资金流向数据失败: {e}")
|
|
data["error"] = str(e)
|
|
|
|
return data
|
|
|
|
def _is_chinese_stock(self, symbol):
|
|
"""判断是否为中国股票"""
|
|
return symbol.isdigit() and len(symbol) == 6
|
|
|
|
def _get_market(self, symbol):
|
|
"""
|
|
根据股票代码判断市场
|
|
上海证券交易所: sh (60开头, 688开头)
|
|
深圳证券交易所: sz (00开头, 30开头)
|
|
北京证券交易所: bj (8开头, 4开头)
|
|
"""
|
|
if symbol.startswith('60') or symbol.startswith('688'):
|
|
return 'sh'
|
|
elif symbol.startswith('00') or symbol.startswith('30'):
|
|
return 'sz'
|
|
elif symbol.startswith('8') or symbol.startswith('4'):
|
|
return 'bj'
|
|
else:
|
|
# 默认深圳
|
|
return 'sz'
|
|
|
|
def _get_individual_fund_flow(self, symbol, market):
|
|
"""获取个股资金流向数据（支持akshare和tushare自动切换）"""
|
|
try:
|
|
# 优先使用tushare
|
|
if data_source_manager.tushare_available:
|
|
try:
|
|
print(f" [Tushare] 正在获取资金流向数据（主要数据源）...")
|
|
ts_code = data_source_manager._convert_to_ts_code(symbol)
|
|
|
|
# 计算日期范围（最近N个交易日）
|
|
end_date = datetime.now().strftime('%Y%m%d')
|
|
start_date = (datetime.now() - timedelta(days=self.days * 2)).strftime('%Y%m%d')
|
|
|
|
df = data_source_manager.tushare_api.moneyflow(
|
|
ts_code=ts_code,
|
|
start_date=start_date,
|
|
end_date=end_date
|
|
)
|
|
|
|
if df is not None and not df.empty:
|
|
# 标准化列名以匹配akshare格式（金额单位：千元→元）
|
|
df = df.rename(columns={
|
|
'trade_date': '日期',
|
|
'close': '收盘价',
|
|
'pct_chg': '涨跌幅',
|
|
'net_mf_amount': '主力净流入-净额'
|
|
})
|
|
df['主力净流入-净额'] = df['主力净流入-净额'] * 1000
|
|
df['超大单净流入-净额'] = (df['buy_elg_amount'] - df['sell_elg_amount']) * 1000
|
|
df['大单净流入-净额'] = (df['buy_lg_amount'] - df['sell_lg_amount']) * 1000
|
|
df['中单净流入-净额'] = (df['buy_md_amount'] - df['sell_md_amount']) * 1000
|
|
df['小单净流入-净额'] = (df['buy_sm_amount'] - df['sell_sm_amount']) * 1000
|
|
# tushare按日期倒序返回，取最近N天后转为正序（与akshare一致）
|
|
df = df.head(self.days)
|
|
df = df.iloc[::-1].reset_index(drop=True)
|
|
print(f" [Tushare] ✅ 成功获取 {len(df)} 条资金流向数据")
|
|
else:
|
|
print(f" [Tushare] ❌ 未找到资金流向数据，尝试备用数据源")
|
|
df = None
|
|
except Exception as te:
|
|
print(f" [Tushare] ❌ 获取失败: {te}")
|
|
df = None
|
|
else:
|
|
df = None
|
|
|
|
# tushare不可用或失败时，回退akshare
|
|
if df is None or df.empty:
|
|
print(f" [Akshare] 正在获取资金流向 (市场: {market})（备用数据源）...")
|
|
df = ak.stock_individual_fund_flow(stock=symbol, market=market)
|
|
if df is None or df.empty:
|
|
print(f" [Akshare] 未找到资金流向数据")
|
|
return None
|
|
|
|
# akshare 返回的数据是按时间正序排列（从旧到新），所以使用 tail() 获取最近N天的数据
|
|
df = df.tail(self.days)
|
|
|
|
# 按日期倒序排列，让最新的数据在前面
|
|
df = df.iloc[::-1].reset_index(drop=True)
|
|
|
|
# 转换为字典列表
|
|
data_list = []
|
|
for idx, row in df.iterrows():
|
|
item = {}
|
|
for col in df.columns:
|
|
value = row.get(col)
|
|
if value is None or (isinstance(value, float) and pd.isna(value)):
|
|
continue
|
|
try:
|
|
# 保持数值类型
|
|
if isinstance(value, (int, float)):
|
|
item[col] = value
|
|
else:
|
|
item[col] = str(value)
|
|
except:
|
|
item[col] = "N/A"
|
|
if item:
|
|
data_list.append(item)
|
|
|
|
return {
|
|
"data": data_list,
|
|
"days": len(data_list),
|
|
"columns": df.columns.tolist(),
|
|
"market": market,
|
|
"query_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f" 获取资金流向数据异常: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return None
|
|
|
|
def format_fund_flow_for_ai(self, data):
|
|
"""
|
|
将资金流向数据格式化为适合AI阅读的文本
|
|
"""
|
|
if not data or not data.get("data_success"):
|
|
return "未能获取资金流向数据"
|
|
|
|
text_parts = []
|
|
|
|
fund_flow_data = data.get("fund_flow_data")
|
|
if fund_flow_data:
|
|
text_parts.append(f"""
|
|
【个股资金流向数据 - akshare数据源】
|
|
股票代码：{data.get('symbol', 'N/A')}
|
|
市场：{fund_flow_data.get('market', 'N/A').upper()}
|
|
交易日数：最近{fund_flow_data.get('days', 0)}个交易日
|
|
查询时间：{fund_flow_data.get('query_time', 'N/A')}
|
|
|
|
═══════════════════════════════════════
|
|
[资金流向详细数据]
|
|
═══════════════════════════════════════
|
|
""")
|
|
|
|
# 显示每个交易日的数据
|
|
for idx, item in enumerate(fund_flow_data.get('data', []), 1):
|
|
date = item.get('日期', 'N/A')
|
|
close_price = item.get('收盘价', 'N/A')
|
|
change_pct = item.get('涨跌幅', 'N/A')
|
|
|
|
text_parts.append(f"""
|
|
第 {idx} 个交易日 ({date}):
|
|
基本信息:
|
|
- 收盘价: {close_price}
|
|
- 涨跌幅: {change_pct}%
|
|
|
|
主力资金:
|
|
- 主力净流入-净额: {item.get('主力净流入-净额', 'N/A')}
|
|
- 主力净流入-净占比: {item.get('主力净流入-净占比', 'N/A')}%
|
|
|
|
超大单:
|
|
- 超大单净流入-净额: {item.get('超大单净流入-净额', 'N/A')}
|
|
- 超大单净流入-净占比: {item.get('超大单净流入-净占比', 'N/A')}%
|
|
|
|
大单:
|
|
- 大单净流入-净额: {item.get('大单净流入-净额', 'N/A')}
|
|
- 大单净流入-净占比: {item.get('大单净流入-净占比', 'N/A')}%
|
|
|
|
中单:
|
|
- 中单净流入-净额: {item.get('中单净流入-净额', 'N/A')}
|
|
- 中单净流入-净占比: {item.get('中单净流入-净占比', 'N/A')}%
|
|
|
|
小单:
|
|
- 小单净流入-净额: {item.get('小单净流入-净额', 'N/A')}
|
|
- 小单净流入-净占比: {item.get('小单净流入-净占比', 'N/A')}%
|
|
""")
|
|
|
|
# 添加统计汇总
|
|
text_parts.append("""
|
|
═══════════════════════════════════════
|
|
[统计汇总 - 最近30个交易日]
|
|
═══════════════════════════════════════
|
|
""")
|
|
|
|
# 计算统计数据
|
|
data_list = fund_flow_data.get('data', [])
|
|
if data_list:
|
|
# 主力净流入统计
|
|
main_inflow_list = [item.get('主力净流入-净额', 0) for item in data_list if isinstance(item.get('主力净流入-净额'), (int, float))]
|
|
if main_inflow_list:
|
|
total_main_inflow = sum(main_inflow_list)
|
|
avg_main_inflow = total_main_inflow / len(main_inflow_list)
|
|
positive_days = len([x for x in main_inflow_list if x > 0])
|
|
negative_days = len([x for x in main_inflow_list if x < 0])
|
|
|
|
text_parts.append(f"""
|
|
主力资金统计:
|
|
- 累计净流入: {total_main_inflow:.2f}
|
|
- 平均每日净流入: {avg_main_inflow:.2f}
|
|
- 净流入天数: {positive_days}天
|
|
- 净流出天数: {negative_days}天
|
|
- 净流入占比: {positive_days/len(main_inflow_list)*100:.1f}%
|
|
""")
|
|
|
|
# 涨跌幅统计
|
|
change_pct_list = [item.get('涨跌幅', 0) for item in data_list if isinstance(item.get('涨跌幅'), (int, float))]
|
|
if change_pct_list:
|
|
avg_change = sum(change_pct_list) / len(change_pct_list)
|
|
up_days = len([x for x in change_pct_list if x > 0])
|
|
down_days = len([x for x in change_pct_list if x < 0])
|
|
|
|
text_parts.append(f"""
|
|
股价统计:
|
|
- 平均涨跌幅: {avg_change:.2f}%
|
|
- 上涨天数: {up_days}天
|
|
- 下跌天数: {down_days}天
|
|
- 上涨占比: {up_days/len(change_pct_list)*100:.1f}%
|
|
""")
|
|
|
|
return "\n".join(text_parts)
|
|
|
|
|
|
# 测试函数
|
|
if __name__ == "__main__":
|
|
print("测试资金流向数据获取（akshare数据源）...")
|
|
print("="*60)
|
|
|
|
fetcher = FundFlowAkshareDataFetcher()
|
|
|
|
if not fetcher.available:
|
|
print("[ERROR] 资金流向数据获取器不可用")
|
|
sys.exit(1)
|
|
|
|
# 测试股票
|
|
test_symbols = [
|
|
("000001", "平安银行"),
|
|
("600519", "贵州茅台"),
|
|
("000858", "五粮液")
|
|
]
|
|
|
|
for symbol, name in test_symbols:
|
|
print(f"\n{'='*60}")
|
|
print(f"正在测试股票: {name} ({symbol})")
|
|
print(f"{'='*60}\n")
|
|
|
|
data = fetcher.get_fund_flow_data(symbol)
|
|
|
|
if data.get("data_success"):
|
|
print("\n" + "="*60)
|
|
print("资金流向数据获取成功！")
|
|
print("="*60)
|
|
|
|
formatted_text = fetcher.format_fund_flow_for_ai(data)
|
|
# 只显示前2000个字符
|
|
preview = formatted_text[:2000] if len(formatted_text) > 2000 else formatted_text
|
|
print(preview)
|
|
if len(formatted_text) > 2000:
|
|
print(f"... (共 {len(formatted_text)} 字符)")
|
|
else:
|
|
print(f"\n获取失败: {data.get('error', '未知错误')}")
|
|
|
|
print("\n")
|