liaolei
/
stock_fundamentals
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

commit;

This commit is contained in:
liao 2025-06-05 10:42:14 +08:00
parent 03450116ce
commit ab27f46c87
14 changed files with 1803 additions and 460 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

729
src/app.py
View File

@ -1,12 +1,12 @@
import sys
import os
from datetime import datetime, timedelta, time
from datetime import datetime, timedelta
import pandas as pd
import uuid
import json
from threading import Thread
from sqlalchemy import create_engine, text
from src.fundamentals_llm.fundamental_analysis_database import get_analysis_result, get_db
from sqlalchemy import text
from src.fundamentals_llm.fundamental_analysis_database import get_db
# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@ -19,7 +19,7 @@ import logging
from src.fundamentals_llm.enterprise_screener import EnterpriseScreener
# 导入股票回测器
from src.stock_analysis_v2 import run_backtest, StockBacktester
from src.stock_analysis_v2 import run_backtest
# 导入PE/PB估值分析器
from src.valuation_analysis.pe_pb_analysis import ValuationAnalyzer
@ -42,9 +42,6 @@ from src.valuation_analysis.index_analyzer import IndexAnalyzer
# 导入股票日线数据采集器
from src.scripts.stock_daily_data_collector import collect_stock_daily_data
from utils.distributed_lock import DistributedLock
from valuation_analysis.industry_analysis import redis_client
from valuation_analysis.financial_analysis import FinancialAnalyzer
from src.valuation_analysis.stock_price_collector import StockPriceCollector
@ -186,224 +183,159 @@ def run_backtest_task(task_id, stocks_buy_dates, end_date):
backtest_tasks[task_id]['error'] = str(e)
logger.error(f"回测任务 {task_id} 失败：{str(e)}")
def initialize_stock_price_schedule():
"""
初始化实时股价数据采集定时任务
"""
# 创建分布式锁
price_lock = DistributedLock(redis_client, "stock_price_collector", expire_time=3600) # 1小时过期
# 尝试获取锁
if not price_lock.acquire():
logger.info("其他服务器正在运行实时股价数据采集任务，本服务器跳过")
return None
@app.route('/scheduler/stockRealtimePrice/collection', methods=['GET'])
def update_stock_realtime_price():
"""更新实时股价数据 周内的9点半、10点半、11点半、2点、3点各更新一次"""
try:
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
# 创建定时任务调度器
scheduler = BackgroundScheduler()
def is_trading_time():
"""判断当前是否为交易时间"""
now = datetime.now()
current_time = now.time()
# 定义交易时间段
morning_start = time(9, 25) # 上午开盘前5分钟
morning_end = time(11, 30) # 上午收盘
afternoon_start = time(13, 0) # 下午开盘
afternoon_end = time(15, 0) # 下午收盘
# 判断是否为工作日
if now.weekday() >= 5: # 5是周六，6是周日
return False
# 判断是否在交易时间段内
is_morning = morning_start <= current_time <= morning_end
is_afternoon = afternoon_start <= current_time <= afternoon_end
return is_morning or is_afternoon
def update_stock_price():
"""更新实时股价数据"""
if not is_trading_time():
return
try:
collector = StockPriceCollector()
collector.update_latest_data()
except Exception as e:
logger.error(f"更新实时股价数据失败: {e}")
# 添加定时任务
scheduler.add_job(
func=update_stock_price,
trigger='interval',
minutes=60,
id='stock_price_update',
name='实时股价数据采集',
replace_existing=True
)
# 启动调度器
scheduler.start()
logger.info("实时股价数据采集定时任务已初始化，将在交易时间内每60分钟执行一次")
return scheduler
collector = StockPriceCollector()
collector.update_latest_data()
except Exception as e:
logger.error(f"初始化实时股价数据采集定时任务失败: {str(e)}")
price_lock.release()
return None
logger.error(f"更新实时股价数据失败: {e}")
return jsonify({
"status": "success"
}), 200
def initialize_rzrq_collector_schedule():
"""初始化融资融券数据采集定时任务"""
# 创建分布式锁
rzrq_lock = DistributedLock(redis_client, "em_rzrq_collector", expire_time=3600) # 1小时过期
# 尝试获取锁
if not rzrq_lock.acquire():
logger.info("其他服务器正在运行融资融券数据采集任务，本服务器跳过")
return None
try:
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
# 创建定时任务调度器
scheduler = BackgroundScheduler()
# 添加每天下午5点执行的任务
scheduler.add_job(
func=run_rzrq_initial_collection,
trigger=CronTrigger(hour=18, minute=40),
id='rzrq_daily_update',
name='每日更新融资融券数据',
replace_existing=True
)
# 启动调度器
scheduler.start()
logger.info("融资融券数据采集定时任务已初始化，将在每天18:00执行")
return scheduler
except Exception as e:
logger.error(f"初始化融资融券数据采集定时任务失败: {str(e)}")
rzrq_lock.release()
return None
def initialize_stock_daily_collector_schedule():
"""初始化股票日线数据采集定时任务"""
# 创建分布式锁
stock_daily_lock = DistributedLock(redis_client, "stock_daily_collector", expire_time=3600) # 1小时过期
# 尝试获取锁
if not stock_daily_lock.acquire():
logger.info("其他服务器正在运行股票日线数据采集任务，本服务器跳过")
return None
try:
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
# 创建定时任务调度器
scheduler = BackgroundScheduler()
# 添加每天下午5点执行的任务
scheduler.add_job(
func=run_stock_daily_collection,
trigger=CronTrigger(hour=15, minute=40),
id='stock_daily_update',
name='每日更新股票日线数据',
replace_existing=True
)
# 启动调度器
scheduler.start()
logger.info("股票日线数据采集定时任务已初始化，将在每天15:40执行")
return scheduler
except Exception as e:
logger.error(f"初始化股票日线数据采集定时任务失败: {str(e)}")
stock_daily_lock.release()
return None
def run_stock_daily_collection():
"""执行股票日线数据采集任务"""
@app.route('/scheduler/stockDaily/collection', methods=['GET'])
def run_stock_daily_collection1():
"""执行股票日线数据采集任务 下午3点四十开始"""
try:
logger.info("开始执行股票日线数据采集")
# 获取当天日期
today = datetime.now().strftime('%Y-%m-%d')
# 定义数据库连接地址
db_url = 'mysql+pymysql://root:Chlry#$.8@192.168.18.199:3306/db_gp_cj'
# 在新线程中执行采集任务，避免阻塞主线程
def collection_task():
try:
# 执行采集
collect_stock_daily_data(db_url, today)
logger.info(f"股票日线数据采集完成，日期: {today}")
except Exception as e:
logger.error(f"执行股票日线数据采集任务失败: {str(e)}")
# 创建并启动线程
thread = Thread(target=collection_task)
thread.daemon = True
thread.start()
return True
collect_stock_daily_data(db_url, today)
except Exception as e:
logger.error(f"启动股票日线数据采集任务失败: {str(e)}")
return False
return jsonify({
"status": "success"
}), 200
def run_rzrq_initial_collection():
"""执行融资融券数据更新采集"""
@app.route('/scheduler/rzrq/collection', methods=['GET'])
def run_rzrq_initial_collection1():
"""执行融资融券数据更新采集 下午7点开始"""
try:
logger.info("开始执行融资融券数据更新采集")
# 生成任务ID
task_id = f"rzrq-{uuid.uuid4().hex[:16]}"
# 记录任务信息
rzrq_tasks[task_id] = {
'status': 'running',
'created_at': datetime.now().isoformat(),
'type': 'initial_collection',
'message': '开始执行融资融券数据更新采集'
}
# 在新线程中执行采集任务
def collection_task():
try:
# 执行采集
result = em_rzrq_collector.initial_data_collection()
if result:
rzrq_tasks[task_id]['status'] = 'completed'
rzrq_tasks[task_id]['message'] = '融资融券数据更新完成'
logger.info(f"融资融券数据更新任务 {task_id} 完成")
else:
rzrq_tasks[task_id]['status'] = 'failed'
rzrq_tasks[task_id]['message'] = '融资融券数据更新失败'
logger.error(f"融资融券数据更新任务 {task_id} 失败")
except Exception as e:
rzrq_tasks[task_id]['status'] = 'failed'
rzrq_tasks[task_id]['message'] = f'执行失败: {str(e)}'
logger.error(f"执行融资融券数据更新线程中出错: {str(e)}")
# 创建并启动线程
thread = Thread(target=collection_task)
thread.daemon = True
thread.start()
return task_id
# 执行采集
em_rzrq_collector.initial_data_collection()
except Exception as e:
logger.error(f"启动融资融券数据更新任务失败: {str(e)}")
if 'task_id' in locals():
rzrq_tasks[task_id]['status'] = 'failed'
rzrq_tasks[task_id]['message'] = f'启动失败: {str(e)}'
return None
return jsonify({
"status": "success"
}), 200
@app.route('/scheduler/industry/crowding', methods=['GET'])
def precalculate_industry_crowding1():
"""预计算部分行业和概念板块的拥挤度指标 晚上10点开始"""
try:
from src.valuation_analysis.industry_analysis import IndustryAnalyzer
analyzer = IndustryAnalyzer()
# 固定行业和概念板块
industries = ["IT设备", "消费电子", "半导体", "军工电子", "专用设备", "乘用车", "产业互联网", "元器件", "光学光电", "医疗器械", "医疗服务", "汽车零部件", "航天装备", "自动化设备"]
concepts = ["先进封装", "芯片", "消费电子概念", "机器人概念"]
# 计算行业拥挤度
for industry in industries:
try:
analyzer.get_industry_crowding_index(industry, use_cache=False)
except Exception as e:
logger.error(f"预计算行业 {industry} 的拥挤度指标时出错: {str(e)}")
continue
# 计算概念板块拥挤度
for concept in concepts:
try:
analyzer.get_industry_crowding_index(concept, use_cache=False, is_concept=True)
except Exception as e:
logger.error(f"预计算概念板块 {concept} 的拥挤度指标时出错: {str(e)}")
continue
logger.info("指定行业和概念板块的拥挤度指标预计算完成")
except Exception as e:
logger.error(f"预计算行业拥挤度指标失败: {str(e)}")
return jsonify({
"status": "success"
}), 200
@app.route('/scheduler/financial/analysis', methods=['GET'])
def scheduler_financial_analysis():
"""预计算所有股票的财务分析数据 早晚各一次"""
try:
from src.valuation_analysis.financial_analysis import FinancialAnalyzer
analyzer = FinancialAnalyzer()
analyzer.analyze_financial_data('601021.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('601021.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('601021.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('600483.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('600483.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('600483.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('688596.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('688596.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('688596.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('002747.SZ', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('002747.SZ', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('002747.SZ', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('688012.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('688012.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('688012.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('603658.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('603658.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('603658.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('002409.SZ', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('002409.SZ', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('002409.SZ', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('600584.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('600584.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('600584.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('603055.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('603055.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('603055.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('601138.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('601138.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('601138.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('603659.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('603659.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('603659.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('688072.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('688072.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('688072.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('688008.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('688008.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('688008.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('300661.SZ', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('300661.SZ', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('300661.SZ', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('603986.SH', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('603986.SH', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('603986.SH', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
analyzer.analyze_financial_data('000733.SZ', current_year = '2024-12-31', previous_year = '2023-12-31', force_update=True)
analyzer.analyze_financial_data('000733.SZ', current_year = '2023-12-31', previous_year = '2022-12-31', force_update=True)
analyzer.analyze_financial_data('000733.SZ', current_year = '2022-12-31', previous_year = '2021-12-31', force_update=True)
except Exception as e:
logger.error(f"预计算所有股票的财务分析数据失败: {str(e)}")
return jsonify({
"status": "success"
}), 200
@app.route('/')
def index():
@ -1551,7 +1483,7 @@ def valuation_analysis():
估值分析接口 - 获取股票的PE/PB估值分析数据
参数:
- stock_code: 股票代码（和stock_name二选一）
- stock_code: 股票代码（支持两种格式：SH601021或601021.SH）
- stock_name: 股票名称（和stock_code二选一）
- start_date: 开始日期（可选，默认为2018-01-01）
- industry_name: 行业名称（可选）
@ -1582,15 +1514,19 @@ def valuation_analysis():
"status": "error",
"message": "请求格式错误: metric参数必须为'pe'或'pb'"
}), 400
# 如果提供了stock_name但没有stock_code，则查询stock_code
if not stock_code and stock_name:
# 这里简化处理，实际项目中应该查询数据库获取股票代码
return jsonify({
"status": "error",
"message": "暂不支持通过股票名称查询，请提供股票代码"
}), 400
# 处理股票代码格式
if stock_code:
if '.' in stock_code: # 处理601021.SH格式
code_parts = stock_code.split('.')
if len(code_parts) == 2:
stock_code = f"{code_parts[1]}{code_parts[0]}"
else:
return jsonify({
"status": "error",
"message": "股票代码格式错误: 应为601021.SH格式"
}), 400
# 验证日期格式
try:
datetime.strptime(start_date, '%Y-%m-%d')
@ -1866,32 +1802,40 @@ def get_concept_list():
@app.route('/api/industry/analysis', methods=['GET'])
def industry_analysis():
"""
行业分析接口 - 获取行业的PE/PB/PS估值分析数据和拥挤度指标
行业/概念板块分析接口 - 获取行业或概念板块的PE/PB/PS估值分析数据和拥挤度指标
参数:
- industry_name: 行业名称
- industry_name: 行业名称（与concept_name二选一）
- concept_name: 概念板块名称（与industry_name二选一）
- metric: 估值指标，可选值为'pe'、'pb'或'ps'，默认为'pe'
- start_date: 开始日期（可选，默认为3年前）
返回:
用于构建ECharts图表的行业估值数据对象，包含估值指标和拥挤度
用于构建ECharts图表的行业/概念板块估值数据对象，包含估值指标和拥挤度
注意：
- 行业PE/PB/PS计算中已剔除负值和极端值(如PE>1000)
- 所有百分位数据都是基于行业平均值计算的
- 行业/概念板块PE/PB/PS计算中已剔除负值和极端值(如PE>1000)
- 所有百分位数据都是基于行业/概念板块平均值计算的
- 拥挤度数据固定使用最近3年的数据，不受start_date参数影响
"""
try:
# 解析参数
industry_name = request.args.get('industry_name')
concept_name = request.args.get('concept_name')
metric = request.args.get('metric', 'pe')
start_date = request.args.get('start_date')
# 检查参数
if not industry_name:
if not industry_name and not concept_name:
return jsonify({
"status": "error",
"message": "请求格式错误: 需要提供industry_name参数"
"message": "请求格式错误: 需要提供industry_name或concept_name参数"
}), 400
if industry_name and concept_name:
return jsonify({
"status": "error",
"message": "请求格式错误: industry_name和concept_name不能同时提供"
}), 400
if metric not in ['pe', 'pb', 'ps']:
@ -1900,8 +1844,13 @@ def industry_analysis():
"message": "请求格式错误: metric参数必须为'pe'、'pb'或'ps'"
}), 400
# 获取行业分析数据
result = industry_analyzer.get_industry_analysis(industry_name, metric, start_date)
# 获取分析数据
if industry_name:
result = industry_analyzer.get_industry_analysis(industry_name, metric, start_date)
title_name = industry_name
else:
result = industry_analyzer.get_concept_analysis(concept_name, metric, start_date)
title_name = concept_name
if not result.get('success', False):
return jsonify({
@ -1918,11 +1867,11 @@ def industry_analysis():
# 准备图例数据
legend_data = [
f"{industry_name}行业平均{metric_name}",
f"行业平均{metric_name}历史最小值",
f"行业平均{metric_name}历史最大值",
f"行业平均{metric_name}历史Q1",
f"行业平均{metric_name}历史Q3"
f"{title_name}平均{metric_name}",
f"平均{metric_name}历史最小值",
f"平均{metric_name}历史最大值",
f"平均{metric_name}历史Q1",
f"平均{metric_name}历史Q3"
]
# 构建结果
@ -1930,7 +1879,7 @@ def industry_analysis():
"status": "success",
"data": {
"title": {
"text": f"{industry_name}行业历史{metric_name}分析",
"text": f"{title_name}历史{metric_name}分析",
"subtext": f"当前{metric_name}百分位: {percentiles['percentile']:.2f}%（剔除负值及极端值）"
},
"tooltip": {
@ -1984,7 +1933,7 @@ def industry_analysis():
],
"series": [
{
"name": f"{industry_name}行业平均{metric_name}",
"name": f"{title_name}平均{metric_name}",
"type": "line",
"data": valuation_data['avg_values'],
"markLine": {
@ -1998,27 +1947,27 @@ def industry_analysis():
}
},
{
"name": f"行业平均{metric_name}历史最小值",
"name": f"平均{metric_name}历史最小值",
"type": "line",
"data": valuation_data['min_values'],
"lineStyle": {"width": 1, "opacity": 0.4, "color": "#28a745"},
"areaStyle": {"opacity": 0.1, "color": "#28a745"}
},
{
"name": f"行业平均{metric_name}历史最大值",
"name": f"平均{metric_name}历史最大值",
"type": "line",
"data": valuation_data['max_values'],
"lineStyle": {"width": 1, "opacity": 0.4, "color": "#dc3545"},
"areaStyle": {"opacity": 0.1, "color": "#dc3545"}
},
{
"name": f"行业平均{metric_name}历史Q1",
"name": f"平均{metric_name}历史Q1",
"type": "line",
"data": valuation_data['q1_values'],
"lineStyle": {"width": 1, "opacity": 0.6, "color": "#28a745"}
},
{
"name": f"行业平均{metric_name}历史Q3",
"name": f"平均{metric_name}历史Q3",
"type": "line",
"data": valuation_data['q3_values'],
"lineStyle": {"width": 1, "opacity": 0.6, "color": "#dc3545"}
@ -2048,7 +1997,7 @@ def industry_analysis():
}
}
# 添加拥挤度指标（如果有）- 作为独立数据，不再添加到主图表series中
# 添加拥挤度指标（如果有）
if "crowding" in result:
crowding_data = result["crowding"]
current_crowding = crowding_data["current"]
@ -2070,7 +2019,7 @@ def industry_analysis():
return jsonify(response)
except Exception as e:
logger.error(f"行业分析请求失败: {str(e)}")
logger.error(f"行业/概念板块分析请求失败: {str(e)}")
return jsonify({
"status": "error",
"message": f"分析失败: {str(e)}"
@ -2091,7 +2040,7 @@ def get_northbound_data():
"""获取北向资金流向数据接口
参数:
- start_time: 可选，开始时间戳（秒）
- start_time: 可选 ，开始时间戳（秒）
- end_time: 可选，结束时间戳（秒）
返回北向资金流向数据
@ -2638,69 +2587,6 @@ def get_index_data():
logger.error(f"获取指数数据失败: {str(e)}")
return jsonify({"status": "error", "message": str(e)})
def initialize_industry_crowding_schedule():
"""初始化行业拥挤度指标预计算定时任务"""
# 创建分布式锁
industry_crowding_lock = DistributedLock(redis_client, "industry_crowding_calculator", expire_time=3600) # 1小时过期
# 尝试获取锁
if not industry_crowding_lock.acquire():
logger.info("其他服务器正在运行行业拥挤度指标预计算任务，本服务器跳过")
return None
try:
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
# 创建定时任务调度器
scheduler = BackgroundScheduler()
# 添加每天晚上10点执行的任务
scheduler.add_job(
func=precalculate_industry_crowding,
trigger=CronTrigger(hour=20, minute=30),
id='industry_crowding_precalc',
name='预计算行业拥挤度指标',
replace_existing=True
)
# 启动调度器
scheduler.start()
logger.info("行业拥挤度指标预计算定时任务已初始化，将在每天20:30执行")
return scheduler
except Exception as e:
logger.error(f"初始化行业拥挤度指标预计算定时任务失败: {str(e)}")
industry_crowding_lock.release()
return None
def precalculate_industry_crowding():
"""预计算所有行业的拥挤度指标"""
try:
from .valuation_analysis.industry_analysis import IndustryAnalyzer
analyzer = IndustryAnalyzer()
industries = analyzer.get_all_industries()
for industry in industries:
try:
# 调用时设置 use_cache=False，强制重新计算
df = analyzer.get_industry_crowding_index(industry, use_cache=False)
if not df.empty:
logger.info(f"成功预计算行业 {industry} 的拥挤度指标")
else:
logger.warning(f"行业 {industry} 的拥挤度指标计算失败")
except Exception as e:
logger.error(f"预计算行业 {industry} 的拥挤度指标时出错: {str(e)}")
continue
logger.info("所有行业的拥挤度指标预计算完成")
except Exception as e:
logger.error(f"预计算行业拥挤度指标失败: {str(e)}")
finally:
# 释放分布式锁
industry_crowding_lock = DistributedLock(redis_client, "industry_crowding_calculator")
industry_crowding_lock.release()
@app.route('/api/financial/analysis', methods=['GET'])
def financial_analysis():
"""
@ -2708,6 +2594,9 @@ def financial_analysis():
请求参数:
stock_code: 股票代码
force_update: 是否强制更新缓存（可选，默认为false）
current_year: 当前年份，格式为'YYYY-12-31'（可选，默认为'2024-12-31'）
previous_year: 上一年份，格式为'YYYY-12-31'（可选，默认为'2023-12-31'）
返回:
分析结果JSON
@ -2719,11 +2608,28 @@ def financial_analysis():
'success': False,
'message': '缺少必要参数：stock_code'
}), 400
analyzer = FinancialAnalyzer()
result = analyzer.analyze_financial_data(stock_code)
result2024 = analyzer.analyze_financial_data(stock_code, current_year = '2024-12-31', previous_year = '2023-12-31')
result2023 = analyzer.analyze_financial_data(stock_code, current_year = '2023-12-31', previous_year = '2022-12-31')
result2022 = analyzer.analyze_financial_data(stock_code, current_year = '2022-12-31', previous_year = '2021-12-31')
# 合并2023和2022的is_better、change_rate、change和avg_score到2024的result
def merge_year_data(target, source, year):
for block in ["financial_strength", "profitability", "growth", "value_rating", "liquidity"]:
if block in target and block in source:
# avg_score
target[block][f"avg_score_{year}"] = source[block].get("avg_score")
# indicators
if "indicators" in target[block] and "indicators" in source[block]:
for t_item, s_item in zip(target[block]["indicators"], source[block]["indicators"]):
for k in ["is_better", "change_rate", "change"]:
if k in s_item:
t_item[f"{k}_{year}"] = s_item.get(k)
merge_year_data(result2024, result2023, "2023")
merge_year_data(result2024, result2022, "2022")
return jsonify(result)
return jsonify(result2024)
except Exception as e:
logger.error(f"财务分析失败: {str(e)}")
@ -2797,45 +2703,178 @@ def test_mongo_structure():
'message': f'测试MongoDB结构失败: {str(e)}'
}), 500
@app.route('/api/stock/real_time_price', methods=['GET'])
def get_real_time_price():
"""获取股票实时价格接口
参数:
- stock_code: 股票代码（必填）
返回:
{
"status": "success",
"data": {
"stock_code": "600000",
"stock_name": "浦发银行",
"current_price": 10.5,
"change_percent": 2.5,
"change_amount": 0.25,
"volume": 1234567,
"amount": 12345678.9,
"high": 10.8,
"low": 10.2,
"open": 10.3,
"pre_close": 10.25,
"update_time": "2024-01-20 14:30:00"
}
}
"""
try:
# 获取股票代码参数
stock_code = request.args.get('stock_code')
# 验证参数
if not stock_code:
return jsonify({
"status": "error",
"message": "缺少必要参数: stock_code"
}), 400
# 导入股票价格采集器
from src.valuation_analysis.stock_price_collector import StockPriceCollector
# 创建采集器实例
collector = StockPriceCollector()
# 获取实时价格数据
price_data = collector.get_stock_price_data(stock_code)
if not price_data:
return jsonify({
"status": "error",
"message": f"获取股票 {stock_code} 的实时价格失败"
}), 404
# 构建响应数据
response_data = {
"stock_code": stock_code,
"stock_name": price_data.get('stock_name'),
"current_price": price_data.get('current_price'),
"change_percent": price_data.get('change_percent'),
"change_amount": price_data.get('change_amount'),
"volume": price_data.get('volume'),
"amount": price_data.get('amount'),
"high": price_data.get('high'),
"low": price_data.get('low'),
"open": price_data.get('open'),
"pre_close": price_data.get('pre_close'),
"update_time": price_data.get('update_time')
}
return jsonify({
"status": "success",
"data": response_data
})
except Exception as e:
logger.error(f"获取股票实时价格异常: {str(e)}")
return jsonify({
"status": "error",
"message": f"服务器错误: {str(e)}"
}), 500
@app.route('/bigscreen')
def bigscreen_page():
"""渲染大屏展示页面"""
return render_template('bigscreen.html')
@app.route('/api/bigscreen_data', methods=['GET'])
def bigscreen_data():
"""聚合大屏所需的12张图数据，便于前端一次性加载"""
try:
# 资金流向
north = hsgt_monitor.fetch_northbound_data()
south = hsgt_monitor.fetch_southbound_data()
# 融资融券
rzrq = em_rzrq_collector.get_chart_data(limit_days=90)
# 恐贪指数
fear_greed = fear_greed_manager.get_index_data(limit=180)
# 概念板块
concepts = [
("先进封装", "xjfz"),
("芯片", "xp"),
("消费电子概念", "xfdz"),
("机器人概念", "jqr")
]
concept_data = {}
for cname, key in concepts:
res = industry_analyzer.get_concept_analysis(cname, 'pe', None)
if res.get('success'):
# PE主线
pe = {
'dates': res['valuation']['dates'],
'values': res['valuation']['avg_values']
}
# 拥挤度
crowding = res.get('crowding', {})
crowding_obj = {
'dates': crowding.get('dates', []),
'values': crowding.get('percentiles', [])
} if crowding else {'dates': [], 'values': []}
concept_data[key] = {'pe': pe, 'crowding': crowding_obj}
else:
concept_data[key] = {'pe': {'dates': [], 'values': []}, 'crowding': {'dates': [], 'values': []}}
return jsonify({
'status': 'success',
'northbound': {
'dates': north.get('times', []),
'values': north.get('data', {}).get('total', [])
} if north.get('success') else {'dates': [], 'values': []},
'southbound': {
'dates': south.get('times', []),
'values': south.get('data', {}).get('total', [])
} if south.get('success') else {'dates': [], 'values': []},
'rzrq': {
'dates': rzrq.get('dates', []),
'values': rzrq.get('series', [{}])[0].get('data', [])
} if rzrq.get('success') and rzrq.get('series') else {'dates': [], 'values': []},
'fear_greed': {
'dates': fear_greed.get('dates', []),
'values': fear_greed.get('values', [])
} if fear_greed.get('success') else {'dates': [], 'values': []},
'concepts': concept_data
})
except Exception as e:
logger.error(f"大屏数据聚合失败: {str(e)}")
return jsonify({'status': 'error', 'message': str(e)})
@app.route('/api/pep_stock_info_by_shortname', methods=['GET'])
def get_pep_stock_info_by_shortname():
"""根据股票简称查询pep_stock_info集合中的全部字段"""
short_name = request.args.get('short_name')
if not short_name:
return jsonify({'success': False, 'message': '缺少必要参数: short_name'}), 400
try:
analyzer = FinancialAnalyzer()
result = analyzer.get_pep_stock_info_by_shortname(short_name)
return jsonify(result)
except Exception as e:
return jsonify({'success': False, 'message': f'服务器错误: {str(e)}'}), 500
@app.route('/api/pep_stock_info_by_code', methods=['GET'])
def get_pep_stock_info_by_code():
"""根据股票简称查询pep_stock_info集合中的全部字段"""
short_code = request.args.get('code')
if not short_code:
return jsonify({'success': False, 'message': '缺少必要参数: short_code'}), 400
try:
analyzer = FinancialAnalyzer()
result = analyzer.get_pep_stock_info_by_code(short_code)
return jsonify(result)
except Exception as e:
return jsonify({'success': False, 'message': f'服务器错误: {str(e)}'}), 500
if __name__ == '__main__':
"""
# 手动释放锁的方法（需要时取消注释）
# 创建锁实例
rzrq_lock = DistributedLock(redis_client, "em_rzrq_collector")
stock_daily_lock = DistributedLock(redis_client, "stock_daily_collector")
industry_crowding_lock = DistributedLock(redis_client, "industry_crowding_calculator")
# 强制释放锁
print("开始释放锁...")
if rzrq_lock.release():
print("成功释放融资融券采集器锁")
else:
print("融资融券采集器锁释放失败或不存在")
if stock_daily_lock.release():
print("成功释放股票日线采集器锁")
else:
print("股票日线采集器锁释放失败或不存在")
if industry_crowding_lock.release():
print("成功释放行业拥挤度锁")
else:
print("行业拥挤度锁释放失败或不存在")
print("锁释放操作完成")
"""
# 初始化融资融券数据采集定时任务
rzrq_scheduler = initialize_rzrq_collector_schedule()
# 初始化股票日线数据采集定时任务
stock_daily_scheduler = initialize_stock_daily_collector_schedule()
# 初始化行业拥挤度指标预计算定时任务
industry_crowding_scheduler = initialize_industry_crowding_schedule()
# 初始化实时股价数据采集定时任务
initialize_stock_price_schedule()
# 启动Web服务器
app.run(host='0.0.0.0', port=5000, debug=True)

5
src/fundamentals_llm/chat_bot.py
View File

@ -132,7 +132,6 @@ class ChatBot:
summary = ref.get('summary', '')
url = ref.get('url', '')
publish_time = ref.get('publish_time', '')
formatted_ref = []
if title:
formatted_ref.append(f"标题：{title}")
@ -150,12 +149,12 @@ class ChatBot:
logger.error(f"格式化参考资料时出错: {str(e)}")
return str(ref)
def chat(self, user_input: str, temperature: float = 1.0, top_p: float = 0.7, max_tokens: int = 4096, frequency_penalty: float = 0.0) -> Dict[str, Any]:
def chat(self, user_input: str, temperature: float = 0.7, top_p: float = 0.7, max_tokens: int = 4096, frequency_penalty: float = 0.0) -> Dict[str, Any]:
"""与AI进行对话
Args:
user_input: 用户输入的问题
temperature: 控制输出的随机性，范围0-2，默认1.0
temperature: 控制输出的随机性，范围0-2，默认0.7
top_p: 控制输出的多样性，范围0-1，默认0.7
max_tokens: 控制输出的最大长度，默认4096
frequency_penalty: 控制重复惩罚，范围-2到2，默认0.0

12
src/fundamentals_llm/fundamental_analysis.py
View File

@ -1464,6 +1464,11 @@ class FundamentalAnalyzer:
if result:
all_results[dimension] = result.ai_response
# 获取当前时间
current_time = datetime.now()
current_year = current_time.year
current_month = current_time.month
# 构建提示词
prompt = f"""请根据以下{stock_name}({stock_code})的各个维度分析结果，生成最终的投资建议，要求输出控制在300字以内，请严格按照以下格式输出：
@ -1477,12 +1482,7 @@ class FundamentalAnalyzer:
- 长期持有：公司具备长期稳定的盈利能力、行业地位稳固、长期成长性好
- 不建议投资：存在明显风险因素、基本面恶化、估值过高、行业前景不佳或者存在退市风险
请注意：
1. 请完全基于提供的分析结果中的最新数据进行分析，不要使用任何历史数据或过时信息
2. 如果分析结果中包含2024年或2025年的数据，请优先使用这些最新数据
3. 避免使用"2023年"等历史时间点的数据，除非分析结果中明确提供了这些数据
4. 重点关注公司最新的业务发展、财务表现和市场定位
5. 在分析行业环境时，请使用最新的行业数据和竞争格局信息
请注意：当前时间是{current_year}年{current_month}月，请基于这个时间点结合当前，分析未来的投资建议。
请提供专业、客观的分析，突出关键信息，避免冗长描述。重点关注投资价值和风险。在输出投资建议时，请明确指出是短期持有、中期持有、长期持有还是不建议投资。

4
src/scripts/config.py
View File

@ -11,7 +11,7 @@ XUEQIU_HEADERS = {
'Accept-Encoding': 'gzip, deflate, br, zstd',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Client-Version': 'v2.44.75',
'Cookie': 'cookiesu=811743062689927; device_id=33fa3c7fca4a65f8f4354e10ed6b7470; HMACCOUNT=8B64A2E3C307C8C0; s=c611ttmqlj; xq_is_login=1; u=8493411634; bid=4065a77ca57a69c83405d6e591ab5449_m8r2nhs8; Hm_lvt_1db88642e346389874251b5a1eded6e3=1746410725; xq_a_token=660fb18cf1d15162da76deedc46b649370124dca; xqat=660fb18cf1d15162da76deedc46b649370124dca; xq_id_token=eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJ1aWQiOjg0OTM0MTE2MzQsImlzcyI6InVjIiwiZXhwIjoxNzQ5ODYxNjY5LCJjdG0iOjE3NDcyNjk2Njk0NDgsImNpZCI6ImQ5ZDBuNEFadXAifQ.jc_E9qvguLwBDASn1Z-KjGtU89pNJRwJq_hIaiR3r2re7-_xiXH8qhuhC3Se8rlfKGZ8sHsb3rSND_vnF7yMp90QQHdK_brSmlgd6_ltHmJfWSFNJvMk7F3s0yPjcpeMqeUTPFnZwKmoWwZVKEwdVBN8f25z6e9M2JjtSTZ2huADH_FdEn1rb9IU-H35z_MLWW1M7vB5xc2rh57yFIBnQoxu9OLfeETpeIpASP1UBeZXoQZ_v1gIWiFYItwuudIz0tPYzB-o2duRe31G0S_hNvEGl3HH4M5FjTyaPAq2PRuiZCyRF-25gHXBZnLcxyavZ1VAURfHng_377_IJNSXsw; xq_r_token=8a5dec9c93caf88d0e1f98f1d23ea1bb60eb6225; snbim_minify=true; is_overseas=0; Hm_lpvt_1db88642e346389874251b5a1eded6e3=1747905410; ssxmod_itna=eqGxBDnGKYuxcD4kDRgxYq7ueYKS8DBP01Dp2xQyP08D60DB40QuHhqDyGGTrlGiGbtOh01qDsqze4GzDiLPGhDBWAFdYCdqtsqfmmxXxyB+doh6odserKO5sg=EiqfqztqpiexCPGnD0=O77N4xYAEDBYD74G+DDeDiO3Dj4GmDGYd=eDFzjRQyl2edxDwDB=DmqG23grDm4DfDDL5xRD4zC2YDDtDAMWz5PDADA3ooDDlYGO44Lr4DYp52nXWdOaspxTXzeDMixGXzYlCgaCRo0TQy9LAN32TNPGuDG=H6e0ahrbicn0AP4KGGwQ0imPKY+5meOQDqixGYwQGGiGGetGe3qqjeKYw10G4ixqim2mpbK+h1iaIPeQAieNS1X5pXZP4rQ04Iv4zmQWvplG40P4Gw4CqRjwzlwGjPwlD3iho+qKlD4hi3YD; ssxmod_itna2=eqGxBDnGKYuxcD4kDRgxYq7ueYKS8DBP01Dp2xQyP08D60DB40QuHhqDyGGTrlGiGbtOh0P4DWhYebouIdHtBItz/DboqtwisfWD',
'Cookie': 'cookiesu=811743062689927; device_id=33fa3c7fca4a65f8f4354e10ed6b7470; smidV2=20250327160437f244626e8b47ca2a7992f30f389e4e790074ae48656a22f10; HMACCOUNT=8B64A2E3C307C8C0; s=c611ttmqlj; xq_is_login=1; u=8493411634; bid=4065a77ca57a69c83405d6e591ab5449_m8r2nhs8; Hm_lvt_1db88642e346389874251b5a1eded6e3=1746410725; __utma=1.434320573.1747189698.1747189698.1747189698.1; __utmc=1; __utmz=1.1747189698.1.1.utmcsr=(direct)|utmccn=(direct)|utmcmd=(none); snbim_minify=true; acw_tc=0a27a9dd17489230816243798e0070441d5e7160c0ed179607143a953db903; xq_a_token=ef79e6da376751a4bf6c1538103e9894d44473e1; xqat=ef79e6da376751a4bf6c1538103e9894d44473e1; xq_id_token=eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJ1aWQiOjg0OTM0MTE2MzQsImlzcyI6InVjIiwiZXhwIjoxNzUxNTE1MDgxLCJjdG0iOjE3NDg5MjMwODE2NDQsImNpZCI6ImQ5ZDBuNEFadXAifQ.gQrIt4VI73JLUFGVSTKpXidhFIMwlusBKyrzYwClwCBszXCooQY3WnFqlbXqSX3SwnMapuveOFUM5sGIOoZ8oDF8cZYs3HDz5vezR-2nes9gfZr2nZcUfZzNRJ299wlX3Zis5NbnzNlfnisUhv9GUfEZjQ_Rs37B4qRbQZVC2kdN1Z0xB8j1MplSTOsYj4IliQntuaTo-8SBh-4zz5244dnF85xREBVxtFzzCtHUhn9B-mzxE81_42nwrDscvow-4_jtlJXlqbehiAFxld-dCWDXwmCju9lRWu_WzdoQe19n-c6jhCZZ1pU1JGsYyhIAsd1gV064jQ6FxfN38so1Eg; xq_r_token=30a80318ebcabffbe194e7deecb108b665e8c894; is_overseas=0; Hm_lpvt_1db88642e346389874251b5a1eded6e3=1748923088; .thumbcache_f24b8bbe5a5934237bbc0eda20c1b6e7=b+jlfRtg2lC80dGHk9izZ9Od1QBbaKrdx1aAMbruXo2ULyhkygsXnhJoa7lOWNgnQAphRKw3864D5K+U2pTL5g%3D%3D; ssxmod_itna=eqGxBDnGKYuxcD4kDRgxYq7ueYKS8DBP01Dp2xQyP08D60DB40Q0shqDyji2YsBGdTYKYUxGXxN4xiNDAc40iDC3WLPeUpx5h5o5Gmxt3qU6P5b48r89Y4sKs=BkpxKFTG4SQW4odeGLDY=DCTKKSMiD4b3Dt4DIDAYDDxDWm4DLDYoDY3uexGPo2mTNpm2bD0YDzqDgD7jbmeDEDG3D0bbetGDo1Q4DGqDSWZHTxD3Dffb4DDN4zIG0GmDDbrR=qmcbC=7O9Wtox0tWDBL5YvysdVC441TXpw8w7WaaxBQD7d9Q5na7fCW13rWkYY0Yeoe7hx+BxYrKch4SbKOAYY7hq7hR0D3E5YD5QADW0D/hQ7Emh07hiY7xdUginMzSTblushiee2YKbK5nYO0t3Ede7d46DqEQMA557QODdNG4WG+slx5bhWiDD; ssxmod_itna2=eqGxBDnGKYuxcD4kDRgxYq7ueYKS8DBP01Dp2xQyP08D60DB40Q0shqDyji2YsBGdTYKY4xDfiOYiiBq4YDj44KWGfmoD/8okGxAeG/0Dt6Q7D6cGudn3qfM5QntBLc5pp/FFY4hly50hUr5qB2v45io/FQi4eD',
'Referer': 'https://weibo.com/u/7735765253',
'Sec-Ch-Ua': '"Chromium";v="122", "Not(A:Brand";v="24", "Google Chrome";v="122"',
'Sec-Ch-Ua-Mobile': '?0',
@ -79,7 +79,7 @@ MODEL_CONFIGS = {
"base_url": "http://192.168.16.174:1234/v1/",
"api_key": "none",
"models": {
"glm-4": "glm-4-32b-0414-abliterated",
"GLM": "glm-4-32b-0414-abliterated",
"qwen3": "qwen3-235b-a22b",
}
},

269
src/static/js/bigscreen.js Normal file
View File

@ -0,0 +1,269 @@
$(function() {
// 1. 资金流向/融资融券/恐贪指数
function fetchNorth() {
$.get('/api/hsgt/northbound', function(res) {
if(res.status === 'success') {
renderNorthChart(res.data);
} else {
$('#northChart').html('<div style="color:#888;text-align:center;padding-top:40px;">交易日9点20采集，暂无数据或数据未更新</div>');
}
}).fail(function() {
$('#northChart').html('<div style="color:#888;text-align:center;padding-top:40px;">交易日9点20采集，暂无数据或数据未更新</div>');
});
}
function fetchSouth() {
$.get('/api/hsgt/southbound', function(res) {
if(res.status === 'success') {
renderSouthChart(res.data);
} else {
$('#southChart').html('<div style="color:#888;text-align:center;padding-top:40px;">交易日9点20采集，暂无数据或数据未更新</div>');
}
}).fail(function() {
$('#southChart').html('<div style="color:#888;text-align:center;padding-top:40px;">交易日9点20采集，暂无数据或数据未更新</div>');
});
}
fetchNorth();
fetchSouth();
setInterval(fetchNorth, 2 * 60 * 1000);
setInterval(fetchSouth, 2 * 60 * 1000);
$.get('/api/rzrq/chart_data?days=90', function(res) {
if(res.status === 'success') renderRzrqChart(res.data);
});
$.get('/api/fear_greed/data?limit=180', function(res) {
if(res.status === 'success') renderFearGreedChart(res.data);
});
// 2. 概念PE和拥挤度
const concepts = [
{name: "先进封装", peId: "peChart_xjfz", crowdId: "crowdChart_xjfz"},
{name: "芯片", peId: "peChart_xp", crowdId: "crowdChart_xp"},
{name: "消费电子概念", peId: "peChart_xfdz", crowdId: "crowdChart_xfdz"},
{name: "机器人概念", peId: "peChart_jqr", crowdId: "crowdChart_jqr"}
];
concepts.forEach(c => {
$.get(`/api/industry/analysis?concept_name=${encodeURIComponent(c.name)}&metric=pe`, function(res) {
if(res.status === 'success') {
renderPEChart(c.peId, res.data);
renderCrowdChart(c.crowdId, res.data.crowding);
}
});
});
// --- 渲染函数 ---
const chartInstances = {};
function renderNorthChart(data) {
if(!data || !data.data || !data.times) return;
const dom = document.getElementById('northChart');
const chart = echarts.init(dom);
chartInstances['northChart'] = chart;
chart.setOption({
title: {text: '', show: false},
tooltip: {trigger: 'axis'},
legend: {data: ['北向资金','沪股通','深股通'], top: 5, textStyle: {color:'#333'}},
grid: {left: '5%', right: '5%', top: 30, bottom: 20, containLabel: true},
xAxis: {type: 'category', data: data.times, axisLabel: {rotate: 0, color:'#666', interval: 'auto'}},
yAxis: {type: 'value', name: '亿元', axisLabel: {color:'#666'}},
series: [
{name: '北向资金', type: 'line', data: data.data.total, symbol: 'none', lineStyle:{width:2}},
{name: '沪股通', type: 'line', data: data.data.sh, symbol: 'none'},
{name: '深股通', type: 'line', data: data.data.sz, symbol: 'none'}
]
});
// 只绑定容器的原生点击事件
dom.onclick = function() {
for(const k in chartInstances) {
if(k !== 'northChart' && chartInstances[k]) {
chartInstances[k].dispatchAction({ type: 'hideTip' });
}
}
};
}
function renderSouthChart(data) {
if(!data || !data.data || !data.times) return;
const dom = document.getElementById('southChart');
const chart = echarts.init(dom);
chartInstances['southChart'] = chart;
chart.setOption({
title: {text: '', show: false},
tooltip: {trigger: 'axis'},
legend: {data: ['南向资金','沪市港股通','深市港股通'], top: 5, textStyle: {color:'#333'}},
grid: {left: '5%', right: '5%', top: 30, bottom: 20, containLabel: true},
xAxis: {type: 'category', data: data.times, axisLabel: {rotate: 0, color:'#666', interval: 'auto'}},
yAxis: {type: 'value', name: '亿元', axisLabel: {color:'#666'}},
series: [
{name: '南向资金', type: 'line', data: data.data.total, symbol: 'none', lineStyle:{width:2}},
{name: '沪市港股通', type: 'line', data: data.data.hk_sh, symbol: 'none'},
{name: '深市港股通', type: 'line', data: data.data.hk_sz, symbol: 'none'}
]
});
dom.onclick = function() {
for(const k in chartInstances) {
if(k !== 'southChart' && chartInstances[k]) {
chartInstances[k].dispatchAction({ type: 'hideTip' });
}
}
};
}
function renderRzrqChart(data) {
if(!data || !data.dates || !data.series) return;
const dom = document.getElementById('rzrqChart');
const chart = echarts.init(dom);
chartInstances['rzrqChart'] = chart;
const s = data.series[0];
let min = Math.min(...s.data.filter(v => v !== null && v !== undefined));
min = Math.floor(min * 0.98);
chart.setOption({
title: {text: '', show: false},
tooltip: {trigger: 'axis'},
legend: {data: [s.name], top: 5, textStyle: {color:'#333'}},
grid: {left: '5%', right: '5%', top: 30, bottom: 20, containLabel: true},
xAxis: {type: 'category', data: data.dates, axisLabel: {rotate: 0, color:'#666', interval: 'auto'}},
yAxis: {type: 'value', name: s.unit, axisLabel: {color:'#666'}, min: min},
series: [{name: s.name, type: 'line', data: s.data, symbol: 'none', lineStyle:{width:2}}]
});
dom.onclick = function() {
for(const k in chartInstances) {
if(k !== 'rzrqChart' && chartInstances[k]) {
chartInstances[k].dispatchAction({ type: 'hideTip' });
}
}
};
}
function renderFearGreedChart(data) {
if(!data || !data.dates || !data.values) return;
const dom = document.getElementById('fearGreedChart');
const chart = echarts.init(dom);
chartInstances['fearGreedChart'] = chart;
let min = Math.min(...data.values.filter(v => v !== null && v !== undefined));
min = Math.floor(min * 0.98);
chart.setOption({
title: {text: '', show: false},
tooltip: {trigger: 'axis'},
legend: {data: ['恐贪指数'], top: 5, textStyle: {color:'#333'}},
grid: {left: '5%', right: '5%', top: 30, bottom: 20, containLabel: true},
xAxis: {type: 'category', data: data.dates, axisLabel: {rotate: 0, color:'#666', interval: 'auto'}},
yAxis: {type: 'value', min: min, max: 100, axisLabel: {color:'#666'}},
series: [{name: '恐贪指数', type: 'line', data: data.values, symbol: 'none', lineStyle:{width:2, color:'#f0ad4e'}}]
});
dom.onclick = function() {
for(const k in chartInstances) {
if(k !== 'fearGreedChart' && chartInstances[k]) {
chartInstances[k].dispatchAction({ type: 'hideTip' });
}
}
};
}
function renderPEChart(domId, data) {
if(!data || !data.series || !data.xAxis || !data.xAxis[0] || !data.xAxis[0].data) return;
const dom = document.getElementById(domId);
const chart = echarts.init(dom);
chartInstances[domId] = chart;
const mainSeries = data.series.filter(s => s.name.indexOf('平均PE') !== -1 || s.name.indexOf('PE') !== -1);
mainSeries.forEach(s => { s.symbol = 'none'; });
let allValues = [];
mainSeries.forEach(s => allValues = allValues.concat(s.data.filter(v => v !== null && v !== undefined)));
let min = Math.min(...allValues);
min = Math.floor(min * 0.98);
chart.setOption({
title: {text: '', show: false},
tooltip: {trigger: 'axis'},
legend: {show: false},
grid: {left: '5%', right: '5%', top: 30, bottom: 20, containLabel: true},
xAxis: {type: 'category', data: data.xAxis[0].data, axisLabel: {rotate: 0, color:'#666', interval: 'auto'}},
yAxis: {type: 'value', name: data.yAxis[0].name, axisLabel: {color:'#666'}, min: min},
series: mainSeries,
dataZoom: [
{type: 'inside', start: 0, end: 100, zoomOnTouch: true, moveOnMouseWheel: true}
]
});
dom.onclick = function() {
for(const k in chartInstances) {
if(k !== domId && chartInstances[k]) {
chartInstances[k].dispatchAction({ type: 'hideTip' });
}
}
};
}
function renderCrowdChart(domId, crowding) {
if(!crowding || !crowding.dates || !crowding.percentiles) return;
// 展示近一年（240天）数据
let dates = crowding.dates;
let percentiles = crowding.percentiles;
if(dates.length > 240) {
dates = dates.slice(-240);
percentiles = percentiles.slice(-240);
}
const dom = document.getElementById(domId);
const chart = echarts.init(dom);
chartInstances[domId] = chart;
let min = Math.min(...percentiles.filter(v => v !== null && v !== undefined));
min = Math.floor(min * 0.98);
// 检查最后一个点是否需要高亮
let markPoint = undefined;
const lastVal = percentiles[percentiles.length-1];
if(lastVal !== undefined && (lastVal > 80 || lastVal < 20)) {
markPoint = {
data: [{
coord: [dates[dates.length-1], lastVal],
symbol: 'circle',
symbolSize: 16,
itemStyle: {
color: lastVal > 80 ? '#ff3333' : '#33cc33',
shadowBlur: 20,
shadowColor: lastVal > 80 ? '#ff3333' : '#33cc33',
opacity: 1
},
label: {show: false},
animation: true,
animationDuration: 500,
animationEasing: 'bounceOut',
animationDurationUpdate: 500,
animationEasingUpdate: 'bounceOut',
effect: {
show: true,
period: 1,
scaleSize: 2,
color: lastVal > 80 ? '#ff3333' : '#33cc33',
shadowBlur: 10
}
}]
};
}
chart.setOption({
title: {text: '', show: false},
tooltip: {trigger: 'axis'},
legend: {data: ['拥挤度历史百分位'], top: 5, textStyle: {color:'#333'}},
grid: {left: '5%', right: '5%', top: 30, bottom: 20, containLabel: true},
xAxis: {type: 'category', data: dates, axisLabel: {rotate: 0, color:'#666', interval: 'auto'}},
yAxis: {type: 'value', min: min, max: 100, name: '百分位(%)', axisLabel: {color:'#666'}},
series: [{
name: '拥挤度历史百分位',
type: 'line',
data: percentiles,
symbol: 'none',
lineStyle:{width:2, color:'#ff7f50'},
markPoint: markPoint
}],
dataZoom: [
{type: 'inside', start: 0, end: 100, zoomOnTouch: true, moveOnMouseWheel: true}
]
});
dom.onclick = function() {
for(const k in chartInstances) {
if(k !== domId && chartInstances[k]) {
chartInstances[k].dispatchAction({ type: 'hideTip' });
}
}
};
}
// 统一resize自适应
$(window).on('resize', function() {
for(const key in chartInstances) {
if(chartInstances[key] && chartInstances[key].resize) {
chartInstances[key].resize();
}
}
});
});

4
src/static/js/echarts.min.js vendored
View File

File diff suppressed because one or more lines are too long

1
src/static/js/hsgt_monitor.js
View File

@ -230,7 +230,6 @@ document.addEventListener('DOMContentLoaded', function() {
? data.data.message
: (data.message || '北向资金数据格式错误');
showError('北向资金数据获取失败: ' + errorMessage);
// 显示空数据状态
renderEmptyNorthboundChart();
}

104
src/static/js/industry.js
View File

@ -6,6 +6,7 @@ document.addEventListener('DOMContentLoaded', function() {
// 获取DOM元素
const industryForm = document.getElementById('industryForm');
const industryNameSelect = document.getElementById('industryName');
const conceptNameSelect = document.getElementById('conceptName');
const startDateInput = document.getElementById('startDate');
const metricSelect = document.getElementById('metric');
const showCrowdingCheckbox = document.getElementById('showCrowding');
@ -28,6 +29,8 @@ document.addEventListener('DOMContentLoaded', function() {
// 初始化 - 加载行业列表
loadIndustryList();
// 加载概念板块列表
loadConceptList();
// 监听表单提交事件
industryForm.addEventListener('submit', function(event) {
@ -113,6 +116,88 @@ document.addEventListener('DOMContentLoaded', function() {
}
}
/**
* 加载概念板块列表
*/
function loadConceptList() {
showLoading(true);
fetch('/api/concept/list')
.then(response => {
if (!response.ok) {
return response.json().then(data => {
throw new Error(data.message || '请求失败');
});
}
return response.json();
})
.then(data => {
if (data.status === 'success') {
// 填充概念板块下拉列表
populateConceptSelect(data.data);
} else {
showError(data.message || '获取概念板块列表失败');
}
})
.catch(error => {
showError(error.message || '请求失败，请检查网络连接');
// 加载失败时使用硬编码的常见概念作为备用
loadFallbackConcepts();
})
.finally(() => {
showLoading(false);
});
}
/**
* 填充概念板块下拉列表
*/
function populateConceptSelect(concepts) {
// 清空选项（保留第一个默认选项）
conceptNameSelect.innerHTML = '<option value="" selected disabled>请选择概念板块</option>';
// 排序概念列表（按名称）
concepts.sort((a, b) => a.name.localeCompare(b.name, 'zh'));
// 添加概念选项
concepts.forEach(concept => {
const option = document.createElement('option');
option.value = concept.name;
option.textContent = concept.name;
conceptNameSelect.appendChild(option);
});
// 如果存在Select2，刷新它
if ($.fn.select2 && $(conceptNameSelect).data('select2')) {
$(conceptNameSelect).trigger('change');
}
}
/**
* 加载备用的概念板块数据（硬编码）
*/
function loadFallbackConcepts() {
const commonConcepts = [
"人工智能", "大数据", "云计算", "物联网", "5G", "新能源", "新材料",
"生物医药", "半导体", "芯片", "消费电子", "智能汽车", "区块链",
"虚拟现实", "元宇宙", "工业互联网", "智能制造", "网络安全", "数字经济"
];
// 先清空下拉框（保留第一个选项）
while (conceptNameSelect.options.length > 1) {
conceptNameSelect.remove(1);
}
// 添加选项
commonConcepts.forEach(concept => {
const option = new Option(concept, concept);
conceptNameSelect.add(option);
});
// 刷新Select2
$(conceptNameSelect).trigger('change');
}
/**
* 分析行业估值
*/
@ -126,11 +211,27 @@ document.addEventListener('DOMContentLoaded', function() {
// 获取表单数据
const industryName = industryNameSelect.value;
const conceptName = conceptNameSelect.value;
const startDate = startDateInput.value;
const metric = metricSelect.value;
// 检查是否至少选择了一个
if (!industryName && !conceptName) {
showError('请选择行业或概念板块');
showLoading(false);
return;
}
// 构建请求URL
let url = `/api/industry/analysis?industry_name=${encodeURIComponent(industryName)}&metric=${metric}`;
let url = `/api/industry/analysis?metric=${metric}`;
if (industryName) {
url += `&industry_name=${encodeURIComponent(industryName)}`;
}
if (conceptName) {
url += `&concept_name=${encodeURIComponent(conceptName)}`;
}
if (startDate) {
url += `&start_date=${startDate}`;
@ -742,6 +843,7 @@ document.addEventListener('DOMContentLoaded', function() {
// 重置Select2
if ($.fn.select2) {
$(industryNameSelect).val('').trigger('change');
$(conceptNameSelect).val('').trigger('change');
}
// 隐藏结果和错误信息

237
src/templates/bigscreen.html Normal file
View File

@ -0,0 +1,237 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>资金与行业估值大屏</title>
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no">
<link rel="stylesheet" href="/static/css/bootstrap.min.css">
<style>
html, body {
height: 100%;
padding-left: 5px;
padding-right: 5px;
padding-top: 5px;
}
body {
background: #f7f7fa;
color: #222;
min-height: 100vh;
}
.container-fluid {
min-height: 100vh;
padding: 0;
}
.row.d-flex {
height: 33vh;
margin-left: 0;
margin-right: 0;
}
.row.d-flex2 {
height: 64vh;
margin-left: 0;
margin-right: 0;
}
.col-3.d-flex {
padding-left: 2px;
padding-right: 2px;
border: 1.5px solid #c7c6c6;
border-radius: 8px;
box-sizing: border-box;
}
.chart-box {
background: #fff;
border-radius: 8px;
padding: 4px 4px 2px 4px;
box-shadow: 0 2px 8px #e0e0e0;
height: 100%;
display: flex;
flex-direction: column;
justify-content: flex-start;
}
.tight-box {
margin: 0;
padding: 2px 2px 1px 2px;
}
.chart-title {
font-size: 0.95rem;
color: #333;
margin-bottom: 2px;
text-align: center;
}
.small-title {
font-size: 0.85rem;
margin-bottom: 2px;
margin-top: 2px;
text-align: center;
color: #666;
}
.chart-container {
width: 100%;
flex: 1 1 0;
min-height: 0;
background: #f9f9fb;
}
/* 平板响应式 */
@media (max-width: 1200px) {
.row.d-flex { height: auto; }
.chart-box { height: auto; min-height: 180px; }
}
/* 移动端响应式 */
@media (max-width: 768px) {
html, body {
padding: 2px;
}
.container-fluid {
padding: 0;
}
.row.d-flex, .row.d-flex2 {
flex-direction: column !important;
height: auto !important;
margin: 0;
}
.col-3.d-flex {
width: 100% !important;
max-width: 100% !important;
min-width: 0 !important;
margin-bottom: 8px;
padding: 1px 0;
}
.chart-box, .tight-box {
min-height: 200px;
height: auto !important;
padding: 2px;
}
.chart-title {
font-size: 0.9rem;
margin-bottom: 1px;
}
.small-title {
font-size: 0.8rem;
margin-bottom: 1px;
margin-top: 1px;
}
.chart-container {
min-height: 180px;
}
/* 调整概念卡片的布局 */
.tight-box .chart-container {
min-height: 150px;
min-width: 0;
width: 100%;
box-sizing: border-box;
}
/* 优化图表间距 */
.tight-box .chart-title {
padding: 1px 0;
}
}
/* 小屏手机响应式 */
@media (max-width: 480px) {
.chart-box, .tight-box {
min-height: 180px;
}
.chart-container {
min-height: 160px;
}
.tight-box .chart-container {
min-height: 130px;
}
.chart-title {
font-size: 0.85rem;
}
.small-title {
font-size: 0.75rem;
}
}
</style>
</head>
<body>
<div class="container-fluid">
<div class="row d-flex">
<div class="col-3 d-flex">
<div class="chart-box w-100">
<button id="fullscreen-btn" style="position:absolute;top:0;left:0;width:100%;height:100%;opacity:0;cursor:pointer;border:none;background:transparent;"></button>
<div class="chart-title">北向资金流向(实时数据)</div>
<div id="northChart" class="chart-container"></div>
</div>
</div>
<div class="col-3 d-flex">
<div class="chart-box w-100">
<div class="chart-title">南向资金流向(实时数据)</div>
<div id="southChart" class="chart-container"></div>
</div>
</div>
<div class="col-3 d-flex">
<div class="chart-box w-100">
<div class="chart-title">融资融券数据监控</div>
<div id="rzrqChart" class="chart-container"></div>
</div>
</div>
<div class="col-3 d-flex">
<div class="chart-box w-100">
<div class="chart-title">市场恐贪指数</div>
<div id="fearGreedChart" class="chart-container"></div>
</div>
</div>
</div>
<div class="row d-flex2">
<div class="col-3 d-flex">
<div class="chart-box tight-box w-100">
<div class="chart-title small-title">先进封装-历史PE分析</div>
<div id="peChart_xjfz" class="chart-container"></div>
<div class="chart-title small-title" style="margin-top:2px;">先进封装-拥挤度</div>
<div id="crowdChart_xjfz" class="chart-container"></div>
</div>
</div>
<div class="col-3 d-flex">
<div class="chart-box tight-box w-100">
<div class="chart-title small-title">芯片-历史PE分析</div>
<div id="peChart_xp" class="chart-container"></div>
<div class="chart-title small-title" style="margin-top:2px;">芯片-拥挤度</div>
<div id="crowdChart_xp" class="chart-container"></div>
</div>
</div>
<div class="col-3 d-flex">
<div class="chart-box tight-box w-100">
<div class="chart-title small-title">消费电子概念-历史PE分析</div>
<div id="peChart_xfdz" class="chart-container"></div>
<div class="chart-title small-title" style="margin-top:2px;">消费电子概念-拥挤度</div>
<div id="crowdChart_xfdz" class="chart-container"></div>
</div>
</div>
<div class="col-3 d-flex">
<div class="chart-box tight-box w-100">
<div class="chart-title small-title">机器人概念-历史PE分析</div>
<div id="peChart_jqr" class="chart-container"></div>
<div class="chart-title small-title" style="margin-top:2px;">机器人概念-拥挤度</div>
<div id="crowdChart_jqr" class="chart-container"></div>
</div>
</div>
</div>
</div>
<script src="/static/js/echarts.min.js"></script>
<script src="/static/js/jquery.min.js"></script>
<script src="/static/js/bigscreen.js"></script>
<script>
// document.getElementById('fullscreen-btn').onclick = function() {
// function launchFullScreen(element) {
// if(element.requestFullscreen) {
// element.requestFullscreen();
// } else if(element.mozRequestFullScreen) {
// element.mozRequestFullScreen();
// } else if(element.webkitRequestFullscreen) {
// element.webkitRequestFullscreen();
// } else if(element.msRequestFullscreen) {
// element.msRequestFullscreen();
// }
// }
// launchFullScreen(document.documentElement);
// this.style.display = 'none'; // 全屏后隐藏按钮
// };
</script>
</body>
</html>

46
src/templates/industry.html
View File

@ -71,6 +71,14 @@
</select>
</div>
<div class="col-md-6">
<label for="conceptName" class="form-label">概念板块</label>
<select class="form-select select2" id="conceptName">
<option value="" selected disabled>请选择概念板块</option>
<!-- 将通过API动态填充 -->
</select>
</div>
<div class="col-md-3">
<label for="metric" class="form-label">估值指标</label>
<select class="form-select" id="metric" required>
@ -195,9 +203,47 @@
width: '100%'
});
// 初始化概念板块下拉框为可搜索
$('#conceptName').select2({
placeholder: '请选择概念板块',
allowClear: true,
width: '100%'
});
// 行业和概念板块互斥选择逻辑
$('#industryName').on('change', function() {
if ($(this).val()) {
// 如果选择了行业，禁用概念板块
$('#conceptName').prop('disabled', true);
// 同时更新Select2的状态
$('#conceptName').select2({disabled: true});
} else {
// 如果清空了行业，启用概念板块
$('#conceptName').prop('disabled', false);
// 同时更新Select2的状态
$('#conceptName').select2({disabled: false});
}
});
// 概念板块选择变化时
$('#conceptName').on('change', function() {
if ($(this).val()) {
// 如果选择了概念，禁用行业
$('#industryName').prop('disabled', true);
// 同时更新Select2的状态
$('#industryName').select2({disabled: true});
} else {
// 如果清空了概念，启用行业
$('#industryName').prop('disabled', false);
// 同时更新Select2的状态
$('#industryName').select2({disabled: false});
}
});
// 重置表单时也需要重置Select2
$('#resetBtn').on('click', function() {
$('#industryName').val('').trigger('change');
$('#conceptName').val('').trigger('change');
});
});
</script>

9
src/valuation_analysis/config.py
View File

@ -26,6 +26,15 @@ MONGO_CONFIG = {
'password': 'wlkj2018',
'collection': 'wind_financial_analysis'
}
# MongoDB配置
MONGO_CONFIG2 = {
'host': '192.168.20.110',
'port': 27017,
'db': 'judge',
'username': 'root',
'password': 'wlkj2018',
'collection': 'wind_financial_analysis'
}
# 项目根目录
ROOT_DIR = Path(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))

107
src/valuation_analysis/eastmoney_rzrq_collector.py
View File

@ -431,15 +431,108 @@ class EastmoneyRzrqCollector:
df = self.fetch_data(page=1)
if not df.empty:
# 保存数据到数据库
if self.save_to_database(df):
logger.info(f"成功更新最新数据，日期：{df.iloc[0]['trade_date']}")
else:
logger.error("更新最新数据失败")
# 确保数据表存在
if not self._ensure_table_exists():
return False
# 处理第一页的所有数据
success_count = 0
total_count = len(df)
with self.engine.connect() as conn:
for _, row in df.iterrows():
# 将Series转换为dict，并处理nan值
row_dict = {k: (None if pd.isna(v) else v) for k, v in row.items()}
# 检查该日期的数据是否已存在
check_query = text("""
SELECT COUNT(*) FROM eastmoney_rzrq_data WHERE trade_date = :trade_date
""")
result = conn.execute(check_query, {"trade_date": row_dict['trade_date']}).scalar()
if result > 0: # 数据已存在，执行更新
update_query = text("""
UPDATE eastmoney_rzrq_data SET
index_value = :index_value,
change_percent = :change_percent,
float_market_value = :float_market_value,
change_percent_3d = :change_percent_3d,
change_percent_5d = :change_percent_5d,
change_percent_10d = :change_percent_10d,
financing_balance = :financing_balance,
financing_balance_ratio = :financing_balance_ratio,
financing_buy_amount = :financing_buy_amount,
financing_buy_amount_3d = :financing_buy_amount_3d,
financing_buy_amount_5d = :financing_buy_amount_5d,
financing_buy_amount_10d = :financing_buy_amount_10d,
financing_repay_amount = :financing_repay_amount,
financing_repay_amount_3d = :financing_repay_amount_3d,
financing_repay_amount_5d = :financing_repay_amount_5d,
financing_repay_amount_10d = :financing_repay_amount_10d,
financing_net_amount = :financing_net_amount,
financing_net_amount_3d = :financing_net_amount_3d,
financing_net_amount_5d = :financing_net_amount_5d,
financing_net_amount_10d = :financing_net_amount_10d,
securities_balance = :securities_balance,
securities_volume = :securities_volume,
securities_repay_volume = :securities_repay_volume,
securities_repay_volume_3d = :securities_repay_volume_3d,
securities_repay_volume_5d = :securities_repay_volume_5d,
securities_repay_volume_10d = :securities_repay_volume_10d,
securities_sell_volume = :securities_sell_volume,
securities_sell_volume_3d = :securities_sell_volume_3d,
securities_sell_volume_5d = :securities_sell_volume_5d,
securities_sell_volume_10d = :securities_sell_volume_10d,
securities_net_volume = :securities_net_volume,
securities_net_volume_3d = :securities_net_volume_3d,
securities_net_volume_5d = :securities_net_volume_5d,
securities_net_volume_10d = :securities_net_volume_10d,
total_rzrq_balance = :total_rzrq_balance,
total_rzrq_balance_cz = :total_rzrq_balance_cz
WHERE trade_date = :trade_date
""")
conn.execute(update_query, row_dict)
logger.info(f"更新数据成功，日期：{row_dict['trade_date']}")
else: # 数据不存在，执行插入
insert_query = text("""
INSERT INTO eastmoney_rzrq_data (
trade_date, index_value, change_percent, float_market_value,
change_percent_3d, change_percent_5d, change_percent_10d,
financing_balance, financing_balance_ratio,
financing_buy_amount, financing_buy_amount_3d, financing_buy_amount_5d, financing_buy_amount_10d,
financing_repay_amount, financing_repay_amount_3d, financing_repay_amount_5d, financing_repay_amount_10d,
financing_net_amount, financing_net_amount_3d, financing_net_amount_5d, financing_net_amount_10d,
securities_balance, securities_volume,
securities_repay_volume, securities_repay_volume_3d, securities_repay_volume_5d, securities_repay_volume_10d,
securities_sell_volume, securities_sell_volume_3d, securities_sell_volume_5d, securities_sell_volume_10d,
securities_net_volume, securities_net_volume_3d, securities_net_volume_5d, securities_net_volume_10d,
total_rzrq_balance, total_rzrq_balance_cz
) VALUES (
:trade_date, :index_value, :change_percent, :float_market_value,
:change_percent_3d, :change_percent_5d, :change_percent_10d,
:financing_balance, :financing_balance_ratio,
:financing_buy_amount, :financing_buy_amount_3d, :financing_buy_amount_5d, :financing_buy_amount_10d,
:financing_repay_amount, :financing_repay_amount_3d, :financing_repay_amount_5d, :financing_repay_amount_10d,
:financing_net_amount, :financing_net_amount_3d, :financing_net_amount_5d, :financing_net_amount_10d,
:securities_balance, :securities_volume,
:securities_repay_volume, :securities_repay_volume_3d, :securities_repay_volume_5d, :securities_repay_volume_10d,
:securities_sell_volume, :securities_sell_volume_3d, :securities_sell_volume_5d, :securities_sell_volume_10d,
:securities_net_volume, :securities_net_volume_3d, :securities_net_volume_5d, :securities_net_volume_10d,
:total_rzrq_balance, :total_rzrq_balance_cz
)
""")
conn.execute(insert_query, row_dict)
logger.info(f"插入新数据成功，日期：{row_dict['trade_date']}")
success_count += 1
conn.commit()
logger.info(f"数据处理完成：成功处理 {success_count}/{total_count} 条记录")
return success_count > 0
else:
logger.warning("未获取到最新数据")
return True
return False
except Exception as e:
logger.error(f"每日更新任务采集失败: {e}")

384
src/valuation_analysis/financial_analysis.py
View File

@ -12,8 +12,9 @@ import logging
from typing import Dict, List, Optional, Union, Tuple
import json
import requests
import redis
from .config import DB_URL, MONGO_CONFIG, LOG_FILE
from .config import DB_URL, MONGO_CONFIG,MONGO_CONFIG2, LOG_FILE
from .stock_price_collector import StockPriceCollector
from .industry_analysis import IndustryAnalyzer
@ -28,6 +29,16 @@ logging.basicConfig(
)
logger = logging.getLogger("financial_analysis")
# 初始化Redis连接
redis_client = redis.Redis(
host='192.168.18.208', # Redis服务器地址，根据实际情况调整
port=6379,
password='wlkj2018',
db=13,
socket_timeout=5,
decode_responses=True
)
class FinancialAnalyzer:
"""财务分析器类"""
@ -48,6 +59,13 @@ class FinancialAnalyzer:
username=MONGO_CONFIG['username'],
password=MONGO_CONFIG['password']
)
# 初始化MongoDB连接
self.mongo_client2 = MongoClient(
host=MONGO_CONFIG2['host'],
port=MONGO_CONFIG2['port'],
username=MONGO_CONFIG2['username'],
password=MONGO_CONFIG2['password']
)
self.mongo_db = self.mongo_client[MONGO_CONFIG['db']]
self.mongo_collection = self.mongo_db[MONGO_CONFIG['collection']]
self.wacc_collection = self.mongo_db['wind_stock_wacc_roic']
@ -97,12 +115,14 @@ class FinancialAnalyzer:
logger.error(f"计算增长变化率失败: {str(e)}")
return None
def get_growth_change_indicators(self, stock_code: str) -> Dict:
def get_growth_change_indicators(self, stock_code: str, current_year: str = '2024-12-31', previous_year: str = '2023-12-31') -> Dict:
"""
获取增长指标的变化率
Args:
stock_code: 股票代码
current_year: 当前年份，格式为'YYYY-12-31'，默认为'2024-12-31'
previous_year: 上一年份，格式为'YYYY-12-31'，默认为'2023-12-31'
Returns:
包含增长指标变化率的字典
@ -116,39 +136,39 @@ class FinancialAnalyzer:
'message': f'未找到股票 {stock_code} 的财务数据'
}
# 获取2023-12-31和2024-12-31的数据
wind_data_2023 = None
wind_data_2024 = None
# 获取指定年份的数据
wind_data_current = None
wind_data_previous = None
for data in record['wind_data']:
if data['time'] == '2023-12-31':
wind_data_2023 = data
elif data['time'] == '2024-12-31':
wind_data_2024 = data
if data['time'] == current_year:
wind_data_current = data
elif data['time'] == previous_year:
wind_data_previous = data
if not wind_data_2023 or not wind_data_2024:
if not wind_data_current or not wind_data_previous:
return {
'success': False,
'message': f'未找到股票 {stock_code} 的2023或2024年财务数据'
'message': f'未找到股票 {stock_code} 的{current_year}或{previous_year}年财务数据'
}
# 获取两个时间点的指标
indicators_2023 = self._get_growth_indicators(wind_data_2023)
indicators_2024 = self._get_growth_indicators(wind_data_2024)
indicators_current = self._get_growth_indicators(wind_data_current)
indicators_previous = self._get_growth_indicators(wind_data_previous)
# 计算变化率
growth_changes = {}
for key in indicators_2023.keys():
current_value = indicators_2024.get(key)
previous_value = indicators_2023.get(key)
for key in indicators_previous.keys():
current_value = indicators_current.get(key)
previous_value = indicators_previous.get(key)
growth_changes[key] = self._calculate_growth_change(current_value, previous_value)
return {
'success': True,
'stock_code': stock_code,
'data': {
'indicators_2023': indicators_2023,
'indicators_2024': indicators_2024,
f'indicators_{previous_year}': indicators_previous,
f'indicators_{current_year}': indicators_current,
'growth_changes': growth_changes
}
}
@ -178,7 +198,7 @@ class FinancialAnalyzer:
})
if not record:
logger.warning(f"未找到股票 {stock_code} 的WACC数据")
# logger.warning(f"未找到股票 {stock_code} 的WACC数据")
return None
return record['wacc']
@ -214,13 +234,13 @@ class FinancialAnalyzer:
logger.error(f"计算盈利年数失败: {str(e)}")
return 0
def extract_financial_indicators(self, stock_code: str) -> Dict:
def extract_financial_indicators(self, stock_code: str, year: str = None) -> Dict:
"""
从MongoDB中提取指定的财务指标
Args:
stock_code: 股票代码
year: 财报年份（可选，格式'YYYY-12-31'），如不传则取最新
Returns:
包含财务指标的字典
"""
@ -232,13 +252,18 @@ class FinancialAnalyzer:
'success': False,
'message': f'未找到股票 {stock_code} 的财务数据'
}
# 获取最新的财务数据（按时间排序）
wind_data = sorted(record['wind_data'], key=lambda x: x['time'], reverse=True)[0]
# 获取指定年份的财务数据（按时间排序）
if year:
wind_data = next((x for x in record['wind_data'] if x['time'] == year), None)
if not wind_data:
return {
'success': False,
'message': f'未找到股票 {stock_code} 的{year}年财务数据'
}
else:
wind_data = sorted(record['wind_data'], key=lambda x: x['time'], reverse=True)[0]
# 获取WACC数据
wacc = self.get_wacc_data(stock_code)
# 计算近五年盈利年数
profit_years = self._calculate_profit_years(record['wind_data'])
@ -458,17 +483,34 @@ class FinancialAnalyzer:
'message': f'获取动量指标失败: {str(e)}'
}
def analyze_financial_data(self, stock_code: str) -> Dict:
def analyze_financial_data(self, stock_code: str, force_update: bool = False, current_year: str = '2024-12-31', previous_year: str = '2023-12-31') -> Dict:
"""
分析财务数据
Args:
stock_code: 股票代码
force_update: 是否强制更新缓存，默认为False
current_year: 当前年份，格式为'YYYY-12-31'，默认为'2024-12-31'
previous_year: 上一年份，格式为'YYYY-12-31'，默认为'2023-12-31'
Returns:
分析结果字典，包含所有财务指标及其排名得分
"""
try:
cache_key = f"financial_analysis:{stock_code}:{current_year}"
# 检查缓存
if not force_update:
cached_data = redis_client.get(cache_key)
if cached_data:
try:
# 尝试解析缓存的JSON数据
result = json.loads(cached_data)
logger.info(f"从缓存获取股票 {stock_code} 的财务分析数据")
return result
except Exception as cache_error:
logger.warning(f"解析缓存的财务分析数据失败，将重新查询: {cache_error}")
# 获取股票价格数据
price_collector = StockPriceCollector()
price_data = price_collector.get_stock_price_data(stock_code)
@ -484,21 +526,45 @@ class FinancialAnalyzer:
momentum_result = self.get_momentum_indicators(stock_code, industry_stocks)
# 获取基础财务指标
base_result = self.extract_financial_indicators(stock_code)
base_result = self.extract_financial_indicators(stock_code, current_year)
if not base_result.get('success'):
return base_result
# 获取增长指标变化
growth_result = self.get_growth_change_indicators(stock_code)
growth_result = self.get_growth_change_indicators(stock_code, current_year, previous_year)
if not growth_result.get('success'):
return growth_result
# 获取行业排名
rank_result = self.calculate_industry_rankings(stock_code)
rank_result = self.calculate_industry_rankings(stock_code, current_year, previous_year)
if not rank_result.get('success'):
return rank_result
# 定义指标说明映射
# 获取指定年份的数据
record = self.mongo_collection.find_one({'code': stock_code})
if not record or 'wind_data' not in record:
return {
'success': False,
'message': f'未找到股票 {stock_code} 的财务数据'
}
# 获取指定年份的数据
wind_data_current = None
wind_data_previous = None
for data in record['wind_data']:
if data['time'] == current_year:
wind_data_current = data
elif data['time'] == previous_year:
wind_data_previous = data
if not wind_data_current or not wind_data_previous:
return {
'success': False,
'message': f'未找到股票 {stock_code} 的{current_year}或{previous_year}年财务数据'
}
# 定义指标说明映射（前端显示用）
indicator_descriptions = {
# 财务实力指标
'debt_equity_ratio': '债务股本比率',
@ -545,10 +611,145 @@ class FinancialAnalyzer:
'cash_ratio': '现金比率'
}
# 定义MongoDB字段映射（用于从MongoDB获取数据）
mongo_field_mapping = {
# 偿债能力指标
'debt_equity_ratio': ('solvency', '产权比率'),
'debt_ebitda_ratio': ('solvency', '全部债务/EBITDA'),
'interest_coverage_ratio': ('solvency', '已获利息倍数(EBIT/利息费用)'),
'current_ratio': ('solvency', '流动比率'),
'quick_ratio': ('solvency', '速动比率'),
'cash_ratio': ('solvency', '现金比率'),
'cash_to_debt_ratio': ('solvency', '经营活动产生的现金流量净额/负债合计'),
# 资本结构指标
'equity_ratio': ('capitalStructure', '股东权益比'),
# 成长能力指标
'diluted_eps_growth': ('grows', '稀释每股收益(同比增长率)'),
'operating_cash_flow_per_share_growth': ('grows', '每股经营活动产生的现金流量净额(同比增长率)'),
'revenue_growth': ('grows', '营业收入(同比增长率)'),
'operating_profit_growth': ('grows', '营业利润(同比增长率)'),
'net_profit_growth_excl_nonrecurring': ('grows', '归属母公司股东的净利润-扣除非经常性损益(同比增长率)'),
'operating_cash_flow_growth': ('grows', '经营活动产生的现金流量净额(同比增长率)'),
'rd_expense_growth': ('grows', '研发费用同比增长'),
'roe_growth': ('grows', '净资产收益率(摊薄)(同比增长)'),
# Z值相关指标
'working_capital_to_assets': ('ZValue', '营运资本/总资产'),
'retained_earnings_to_assets': ('ZValue', '留存收益/总资产'),
'ebit_to_assets': ('ZValue', '息税前利润(TTM)/总资产'),
'market_value_to_liabilities': ('ZValue', '当日总市值/负债总计'),
'equity_to_liabilities': ('ZValue', '股东权益合计(含少数)/负债总计'),
'revenue_to_assets': ('ZValue', '营业收入/总资产'),
'z_score': ('ZValue', 'Z值'),
# 营运能力指标
'inventory_turnover_days': ('operatingCapacity', '存货周转天数'),
'receivables_turnover_days': ('operatingCapacity', '应收账款周转天数'),
'payables_turnover_days': ('operatingCapacity', '应付账款周转天数'),
# 盈利能力指标
'gross_profit_margin': ('profitability', '销售毛利率'),
'operating_profit_margin': ('profitability', '营业利润/营业总收入'),
'net_profit_margin': ('profitability', '销售净利率'),
'roe': ('profitability', '净资产收益率ROE(平均)'),
'roa': ('profitability', '总资产净利率ROA'),
'roic': ('profitability', '投入资本回报率ROIC')
}
# 定义指标是否越高越好
higher_better_indicators = {
# 偿债能力指标
'current_ratio': True,
'quick_ratio': True,
'cash_ratio': True,
'interest_coverage_ratio': True,
# 资本结构指标
'equity_ratio': True,
# 盈利能力指标
'profit_years': True,
# 成长能力指标
'diluted_eps_growth': True,
'operating_cash_flow_per_share_growth': True,
'revenue_growth': True,
'operating_profit_growth': True,
'net_profit_growth_excl_nonrecurring': True,
'operating_cash_flow_growth': True,
'rd_expense_growth': True,
'roe_growth': True,
# Z值相关指标
'working_capital_to_assets': True,
'retained_earnings_to_assets': True,
'ebit_to_assets': True,
'market_value_to_liabilities': True,
'equity_to_liabilities': True,
'revenue_to_assets': True,
'z_score': True,
# 营运能力指标
'inventory_turnover_days': False,
'receivables_turnover_days': False,
'payables_turnover_days': False,
# 盈利能力指标
'gross_profit_margin': True,
'operating_profit_margin': True,
'net_profit_margin': True,
'roe': True,
'roa': True,
'roic': True,
# WACC数据
'wacc': False
}
# 构建基础指标数据
base_indicators = base_result.get('indicators', {})
rankings = rank_result.get('rankings', {})
def get_indicator_value(data: Dict, key: str) -> Optional[float]:
"""获取指定指标的值"""
if key not in mongo_field_mapping:
return None
category, meaning = mongo_field_mapping[key]
if category not in data:
return None
for item in data[category]['list']:
if item['meaning'] == meaning:
return item['data']
return None
def calculate_change(current: Optional[float], previous: Optional[float], is_higher_better: bool) -> Dict:
"""计算指标变化情况"""
if current is None or previous is None:
return {
'change': None,
'change_rate': None,
'is_better': None
}
change = current - previous
change_rate = (change / abs(previous)) * 100 if previous != 0 else None
# 判断是否变好
if is_higher_better:
is_better = change > 0
else:
is_better = change < 0
return {
'change': round(change, 3),
'change_rate': round(change_rate, 2) if change_rate is not None else None,
'is_better': is_better
}
# 定义各板块的指标列表
financial_strength_indicators = [
'debt_equity_ratio', 'debt_ebitda_ratio', 'interest_coverage_ratio',
@ -578,7 +779,6 @@ class FinancialAnalyzer:
'payables_turnover_days', 'current_ratio', 'quick_ratio', 'cash_ratio'
]
# 处理各板块指标
def process_indicators(indicator_list):
result = []
total_score = 0
@ -586,6 +786,13 @@ class FinancialAnalyzer:
for key in indicator_list:
if key in base_indicators:
current_value = base_indicators[key]
previous_value = get_indicator_value(wind_data_previous, key)
# 计算变化情况
is_higher_better = higher_better_indicators.get(key, True)
change_info = calculate_change(current_value, previous_value, is_higher_better)
rank_score = rankings.get(key, 0)
if rank_score is not None:
total_score += rank_score
@ -594,7 +801,11 @@ class FinancialAnalyzer:
result.append({
'key': key,
'desc': indicator_descriptions.get(key, key),
'value': base_indicators[key],
'value': current_value,
'previous_value': previous_value,
'change': change_info['change'],
'change_rate': change_info['change_rate'],
'is_better': change_info['is_better'],
'rank_score': rank_score
})
@ -606,6 +817,12 @@ class FinancialAnalyzer:
'avg_score': avg_score
}
# 处理各板块指标
financial_strength = process_indicators(financial_strength_indicators)
profitability = process_indicators(profitability_indicators)
value_rating = process_indicators(value_rating_indicators)
liquidity = process_indicators(liquidity_indicators)
# 处理增长指标变化
growth_changes = growth_result.get('data', {}).get('growth_changes', {})
growth_changes_list = []
@ -639,20 +856,38 @@ class FinancialAnalyzer:
'avg_score': growth_avg_score
}
return {
industryList = industry_analyzer.get_stock_industry(stock_code)
concepts += industryList
# 在返回结果之前，缓存数据
result = {
'success': True,
'stock_code': stock_code,
'data_time': base_indicators.get('data_time'),
'financial_strength': process_indicators(financial_strength_indicators),
'profitability': process_indicators(profitability_indicators),
'financial_strength': financial_strength,
'profitability': profitability,
'growth': growth_data,
'value_rating': process_indicators(value_rating_indicators),
'liquidity': process_indicators(liquidity_indicators),
'momentum': momentum_result.get('indicators', []), # 添加动量指标数据
'concepts': concepts, # 添加概念板块数据
'price_data': price_data # 添加实时股价数据
'value_rating': value_rating,
'liquidity': liquidity,
'momentum': momentum_result.get('indicators', []),
'concepts': concepts,
'price_data': price_data
}
# 缓存结果，有效期1天（86400秒）
try:
redis_client.set(
cache_key,
json.dumps(result, default=str), # 使用default=str处理日期等特殊类型
ex=86400 # 1天的秒数
)
logger.info(f"已缓存股票 {stock_code} 的财务分析数据，有效期为1天")
except Exception as cache_error:
logger.warning(f"缓存财务分析数据失败: {cache_error}")
return result
except Exception as e:
logger.error(f"分析财务数据失败: {str(e)}")
return {
@ -805,13 +1040,14 @@ class FinancialAnalyzer:
logger.error(f"获取行业指标失败: {str(e)}")
return {}
def calculate_industry_rankings(self, stock_code: str) -> Dict:
def calculate_industry_rankings(self, stock_code: str, current_year: str = '2024-12-31', previous_year: str = '2023-12-31') -> Dict:
"""
计算公司在行业中的排名得分
Args:
stock_code: 股票代码
current_year: 当前年份，格式为'YYYY-12-31'，默认为'2024-12-31'
previous_year: 上一年份，格式为'YYYY-12-31'，默认为'2023-12-31'
Returns:
包含所有指标排名得分的字典
"""
@ -825,22 +1061,32 @@ class FinancialAnalyzer:
}
# 获取当前公司的指标
current_result = self.extract_financial_indicators(stock_code)
current_result = self.extract_financial_indicators(stock_code, year=current_year)
if not current_result.get('success'):
return current_result
# 获取当前公司的增长指标变化
current_growth_result = self.get_growth_change_indicators(stock_code)
current_growth_result = self.get_growth_change_indicators(stock_code, current_year, previous_year)
if not current_growth_result.get('success'):
return current_growth_result
# 获取行业所有公司的指标
industry_indicators = self._get_industry_indicators(stock_list)
industry_indicators = {}
for stock in stock_list:
result = self.extract_financial_indicators(stock, year=current_year)
if not result.get('success'):
continue
indicators = result.get('indicators', {})
for key, value in indicators.items():
if key != 'data_time':
if key not in industry_indicators:
industry_indicators[key] = []
industry_indicators[key].append(value)
# 获取行业所有公司的增长指标变化
industry_growth_indicators = {}
for stock in stock_list:
growth_result = self.get_growth_change_indicators(stock)
growth_result = self.get_growth_change_indicators(stock, current_year, previous_year)
if growth_result.get('success'):
growth_changes = growth_result.get('data', {}).get('growth_changes', {})
for key, value in growth_changes.items():
@ -937,4 +1183,46 @@ class FinancialAnalyzer:
return {
'success': False,
'message': f'计算行业排名失败: {str(e)}'
}
}
def get_pep_stock_info_by_shortname(self, short_name: str) -> dict:
"""
根据股票简称从pep_stock_info集合获取对应的全部字段数据
Args:
short_name: 股票简称
Returns:
查询到的文档（dict），未找到则返回{'success': False, 'message': ...}
"""
try:
pep_collection = self.mongo_client[MONGO_CONFIG2['db']]['pep_stock_info']
record = pep_collection.find_one({'shortName': short_name})
if not record:
return {'success': False, 'message': f'未找到简称为{short_name}的股票数据'}
if '_id' in record:
record['_id'] = str(record['_id']) # 转为字符串，便于JSON序列化
record['success'] = True
return record
except Exception as e:
logger.error(f"查询pep_stock_info失败: {str(e)}")
return {'success': False, 'message': f'查询pep_stock_info失败: {str(e)}'}
def get_pep_stock_info_by_code(self, short_code: str) -> dict:
"""
根据股票简称从pep_stock_info集合获取对 应的全部字段数据
Args:
short_code: 股票简称
Returns:
查询到的文档（dict），未找到则返回{'success': False, 'message': ...}
"""
try:
pep_collection = self.mongo_client2[MONGO_CONFIG2['db']]['pep_stock_info']
record = pep_collection.find_one({'code': short_code})
if not record:
return {'success': False, 'message': f'未找到简称为{short_code}的股票数据'}
if '_id' in record:
record['_id'] = str(record['_id']) # 转为字符串，便于JSON序列化
record['success'] = True
return record
except Exception as e:
logger.error(f"查询pep_stock_info失败: {str(e)}")
return {'success': False, 'message': f'查询pep_stock_info失败: {str(e)}'}

352
src/valuation_analysis/industry_analysis.py
View File

@ -274,56 +274,56 @@ class IndustryAnalyzer:
logger.info(f"计算行业 {metric} 分位数完成: 当前{metric}={result['current']:.2f}, 百分位={result['percentile']:.2f}%")
return result
def get_industry_crowding_index(self, industry_name: str, start_date: str = None, end_date: str = None, use_cache: bool = True) -> pd.DataFrame:
def get_industry_crowding_index(self, name: str, start_date: str = None, end_date: str = None, use_cache: bool = True, is_concept: bool = False) -> pd.DataFrame:
"""
计算行业交易拥挤度指标，并使用Redis缓存结果
计算行业/概念板块交易拥挤度指标，并使用Redis缓存结果
对于拥挤度指标，固定使用3年数据，不受start_date影响
缓存时间为1天
Args:
industry_name: 行业名称
name: 行业/概念板块名称
start_date: 不再使用此参数，保留是为了兼容性
end_date: 结束日期（默认为当前日期）
use_cache: 是否使用缓存，默认为True
is_concept: 是否为概念板块，默认为False
Returns:
包含行业拥挤度指标的DataFrame
包含行业/概念板块拥挤度指标的DataFrame
"""
try:
# 始终使用3年前作为开始日期
three_years_ago = (datetime.datetime.now() - datetime.timedelta(days=3*365)).strftime('%Y-%m-%d')
cache_key = f"{'concept' if is_concept else 'industry'}_crowding:{name}"
if end_date is None:
end_date = datetime.datetime.now().strftime('%Y-%m-%d')
# 检查缓存
if use_cache:
cache_key = f"industry_crowding:{industry_name}"
cached_data = redis_client.get(cache_key)
if cached_data:
try:
# 尝试解析缓存的JSON数据
cached_df_dict = json.loads(cached_data)
logger.info(f"从缓存获取行业 {industry_name} 的拥挤度数据")
# 将缓存的字典转换回DataFrame
df = pd.DataFrame(cached_df_dict)
# 确保trade_date列是日期类型
df['trade_date'] = pd.to_datetime(df['trade_date'])
return df
except Exception as cache_error:
logger.warning(f"解析缓存的拥挤度数据失败，将重新查询: {cache_error}")
# 获取行业所有股票
stock_codes = self.get_industry_stocks(industry_name)
cached_data = redis_client.get(cache_key)
if cached_data:
try:
# 尝试解析缓存的JSON数据
cached_df_dict = json.loads(cached_data)
logger.info(f"从缓存获取{'概念板块' if is_concept else '行业'} {name} 的拥挤度数据")
# 将缓存的字典转换回DataFrame
df = pd.DataFrame(cached_df_dict)
# 确保trade_date列是日期类型
df['trade_date'] = pd.to_datetime(df['trade_date'])
return df
except Exception as cache_error:
logger.warning(f"解析缓存的拥挤度数据失败，将重新查询: {cache_error}")
# 获取行业/概念板块所有股票
stock_codes = self.get_concept_stocks(name) if is_concept else self.get_industry_stocks(name)
if not stock_codes:
return pd.DataFrame()
# 优化方案：分别查询市场总成交额和行业成交额，然后在Python中计算比率
# 优化方案：分别查询市场总成交额和行业/概念板块成交额，然后在Python中计算比率
# 查询1：获取每日总成交额
query_total = text("""
@ -340,11 +340,11 @@ class IndustryAnalyzer:
`timestamp`
""")
# 查询2：获取行业每日成交额
query_industry = text("""
# 查询2：获取行业/概念板块每日成交额
query_sector = text("""
SELECT
`timestamp` AS trade_date,
SUM(amount) AS industry_amount
SUM(amount) AS sector_amount
FROM
gp_day_data
WHERE
@ -364,8 +364,8 @@ class IndustryAnalyzer:
params={"start_date": three_years_ago, "end_date": end_date}
)
df_industry = pd.read_sql(
query_industry,
df_sector = pd.read_sql(
query_sector,
conn,
params={
"stock_codes": tuple(stock_codes),
@ -375,15 +375,15 @@ class IndustryAnalyzer:
)
# 检查查询结果
if df_total.empty or df_industry.empty:
logger.warning(f"未找到行业 {industry_name} 的交易数据")
if df_total.empty or df_sector.empty:
logger.warning(f"未找到{'概念板块' if is_concept else '行业'} {name} 的交易数据")
return pd.DataFrame()
# 在Python中合并数据并计算比率
df = pd.merge(df_total, df_industry, on='trade_date', how='inner')
df = pd.merge(df_total, df_sector, on='trade_date', how='inner')
# 计算行业成交额占比
df['industry_amount_ratio'] = (df['industry_amount'] / df['total_market_amount']) * 100
# 计算行业/概念板块成交额占比
df['industry_amount_ratio'] = (df['sector_amount'] / df['total_market_amount']) * 100
# 在Python中计算百分位
df['percentile'] = df['industry_amount_ratio'].rank(pct=True) * 100
@ -399,22 +399,21 @@ class IndustryAnalyzer:
df_dict = df.to_dict(orient='records')
# 缓存结果，有效期1天（86400秒）
if use_cache:
try:
redis_client.set(
cache_key,
json.dumps(df_dict, default=str), # 使用default=str处理日期等特殊类型
ex=86400 # 1天的秒数
)
logger.info(f"已缓存行业 {industry_name} 的拥挤度数据，有效期为1天")
logger.info(f"已缓存{'概念板块' if is_concept else '行业'} {name} 的拥挤度数据，有效期为1天")
except Exception as cache_error:
logger.warning(f"缓存行业拥挤度数据失败: {cache_error}")
logger.info(f"成功计算行业 {industry_name} 的拥挤度指标，共 {len(df)} 条记录")
logger.warning(f"缓存拥挤度数据失败: {cache_error}")
logger.info(f"成功计算{'概念板块' if is_concept else '行业'} {name} 的拥挤度指标，共 {len(df)} 条记录")
return df
except Exception as e:
logger.error(f"计算行业拥挤度指标失败: {e}")
logger.error(f"计算拥挤度指标失败: {e}")
return pd.DataFrame()
def get_industry_analysis(self, industry_name: str, metric: str = 'pe', start_date: str = None) -> Dict:
@ -508,11 +507,25 @@ class IndustryAnalyzer:
}
}
# ========== 新增：对所有数值四舍五入到两位小数 ==========
def round_floats(obj):
if isinstance(obj, float):
return round(obj, 2)
elif isinstance(obj, int):
return obj # int不处理，保留原样（如有需要可加round）
elif isinstance(obj, list):
return [round_floats(x) for x in obj]
elif isinstance(obj, dict):
return {k: round_floats(v) for k, v in obj.items()}
else:
return obj
result = round_floats(result)
# =====================================================
return result
except Exception as e:
logger.error(f"获取行业综合分析失败: {e}")
return {"success": False, "message": f"获取行业综合分析失败: {e}"}
return {"success": False, "message": f"获取行业综合分析失败: {e}"}
def get_stock_concepts(self, stock_code: str) -> List[str]:
"""
@ -547,6 +560,39 @@ class IndustryAnalyzer:
logger.error(f"获取股票概念板块失败: {e}")
return []
def get_stock_industry(self, stock_code: str) -> List[str]:
"""
获取指定股票所属的概念板块列表
Args:
stock_code: 股票代码
Returns:
概念板块名称列表
"""
try:
# 转换股票代码格式
formatted_code = self._convert_stock_code_format(stock_code)
query = text("""
SELECT DISTINCT bk_name
FROM gp_hybk
WHERE gp_code = :stock_code
""")
with self.engine.connect() as conn:
result = conn.execute(query, {"stock_code": formatted_code}).fetchall()
if result:
return [row[0] for row in result]
else:
logger.warning(f"未找到股票 {stock_code} 的行业板块数据")
return []
except Exception as e:
logger.error(f"获取股票行业板块失败: {e}")
return []
def _convert_stock_code_format(self, stock_code: str) -> str:
"""
转换股票代码格式
@ -562,4 +608,220 @@ class IndustryAnalyzer:
return f"{market}{code}"
except Exception as e:
logger.error(f"转换股票代码格式失败: {str(e)}")
return stock_code
return stock_code
def get_concept_stocks(self, concept_name: str) -> List[str]:
"""
获取指定概念板块的所有股票代码
Args:
concept_name: 概念板块名称
Returns:
股票代码列表
"""
try:
query = text("""
SELECT DISTINCT gp_code
FROM gp_gnbk
WHERE bk_name = :concept_name
""")
with self.engine.connect() as conn:
result = conn.execute(query, {"concept_name": concept_name}).fetchall()
if result:
return [row[0] for row in result]
else:
logger.warning(f"未找到概念板块 {concept_name} 的股票")
return []
except Exception as e:
logger.error(f"获取概念板块股票失败: {e}")
return []
def get_concept_valuation_data(self, concept_name: str, start_date: str, metric: str = 'pe') -> pd.DataFrame:
"""
获取概念板块估值数据，返回每日概念板块平均PE/PB/PS
说明：
- 概念板块估值数据是指概念板块内所有股票的平均PE/PB/PS的历史数据
- 在计算过程中会剔除负值和极端值(如PE>1000)
Args:
concept_name: 概念板块名称
start_date: 开始日期
metric: 估值指标（pe、pb或ps）
Returns:
包含概念板块估值数据的DataFrame，主要包含以下列：
- timestamp: 日期
- avg_{metric}: 概念板块平均值
- stock_count: 参与计算的股票数量
"""
try:
# 验证metric参数
if metric not in ['pe', 'pb', 'ps']:
logger.error(f"不支持的估值指标: {metric}")
return pd.DataFrame()
# 获取概念板块所有股票
stock_codes = self.get_concept_stocks(concept_name)
if not stock_codes:
return pd.DataFrame()
# 构建查询 - 只计算每天的概念板块平均值和参与计算的股票数量
query = text(f"""
WITH valid_data AS (
SELECT
`timestamp`,
symbol,
{metric}
FROM
gp_day_data
WHERE
symbol IN :stock_codes AND
`timestamp` >= :start_date AND
{metric} > 0 AND
{metric} < 1000 -- 过滤掉极端异常值
)
SELECT
`timestamp`,
AVG({metric}) as avg_{metric},
COUNT(*) as stock_count
FROM
valid_data
GROUP BY
`timestamp`
ORDER BY
`timestamp`
""")
with self.engine.connect() as conn:
# 获取汇总数据
df = pd.read_sql(
query,
conn,
params={"stock_codes": tuple(stock_codes), "start_date": start_date}
)
if df.empty:
logger.warning(f"未找到概念板块 {concept_name} 的估值数据")
return pd.DataFrame()
logger.info(f"成功获取概念板块 {concept_name} 的{metric.upper()}数据，共 {len(df)} 条记录")
return df
except Exception as e:
logger.error(f"获取概念板块估值数据失败: {e}")
return pd.DataFrame()
def get_concept_analysis(self, concept_name: str, metric: str = 'pe', start_date: str = None) -> Dict:
"""
获取概念板块综合分析结果
Args:
concept_name: 概念板块名称
metric: 估值指标（pe、pb或ps）
start_date: 开始日期（默认为3年前）
Returns:
概念板块分析结果字典，包含以下内容：
- success: 是否成功
- concept_name: 概念板块名称
- metric: 估值指标
- analysis_date: 分析日期
- valuation: 估值数据，包含：
- dates: 日期列表
- avg_values: 概念板块平均值列表
- stock_counts: 参与计算的股票数量列表
- percentiles: 分位数信息，包含概念板块平均值的历史最大值、最小值、四分位数等
- crowding（如有）: 拥挤度数据，包含：
- dates: 日期列表
- ratios: 拥挤度比例列表
- percentiles: 拥挤度百分位列表
- current: 当前拥挤度信息
"""
try:
# 默认查询近3年数据
if start_date is None:
start_date = (datetime.datetime.now() - datetime.timedelta(days=3*365)).strftime('%Y-%m-%d')
# 获取估值数据
valuation_data = self.get_concept_valuation_data(concept_name, start_date, metric)
if valuation_data.empty:
return {"success": False, "message": f"无法获取概念板块 {concept_name} 的估值数据"}
# 计算估值分位数
percentiles = self.calculate_industry_percentiles(valuation_data, metric)
if not percentiles:
return {"success": False, "message": f"无法计算概念板块 {concept_name} 的估值分位数"}
# 获取拥挤度指标（始终使用3年数据，不受start_date影响）
crowding_data = self.get_industry_crowding_index(concept_name, is_concept=True)
# 为了兼容前端，准备一些概念板块平均值的历史统计数据
avg_values = valuation_data[f'avg_{metric}'].tolist()
# 准备返回结果
result = {
"success": True,
"concept_name": concept_name,
"metric": metric.upper(),
"analysis_date": datetime.datetime.now().strftime('%Y-%m-%d'),
"valuation": {
"dates": valuation_data['timestamp'].dt.strftime('%Y-%m-%d').tolist(),
"avg_values": avg_values,
# 填充概念板块平均值的历史统计线
"min_values": [percentiles['min']] * len(avg_values), # 概念板块平均PE历史最小值
"max_values": [percentiles['max']] * len(avg_values), # 概念板块平均PE历史最大值
"q1_values": [percentiles['q1']] * len(avg_values), # 概念板块平均PE历史第一四分位数
"q3_values": [percentiles['q3']] * len(avg_values), # 概念板块平均PE历史第三四分位数
"median_values": [percentiles['median']] * len(avg_values), # 概念板块平均PE历史中位数
"stock_counts": valuation_data['stock_count'].tolist(),
"percentiles": percentiles
}
}
# 添加拥挤度数据（如果有）
if not crowding_data.empty:
current_crowding = crowding_data.iloc[-1]
result["crowding"] = {
"dates": crowding_data['trade_date'].dt.strftime('%Y-%m-%d').tolist(),
"ratios": crowding_data['industry_amount_ratio'].tolist(),
"percentiles": crowding_data['percentile'].tolist(),
"current": {
"date": current_crowding['trade_date'].strftime('%Y-%m-%d'),
"ratio": float(current_crowding['industry_amount_ratio']),
"percentile": float(current_crowding['percentile']),
"level": current_crowding['crowding_level'],
# 添加概念板块成交额比例的历史分位信息
"ratio_stats": {
"min": float(crowding_data['industry_amount_ratio'].min()),
"max": float(crowding_data['industry_amount_ratio'].max()),
"mean": float(crowding_data['industry_amount_ratio'].mean()),
"median": float(crowding_data['industry_amount_ratio'].median()),
"q1": float(crowding_data['industry_amount_ratio'].quantile(0.25)),
"q3": float(crowding_data['industry_amount_ratio'].quantile(0.75)),
}
}
}
# ========== 新增：对所有数值四舍五入到两位小数 ==========
def round_floats(obj):
if isinstance(obj, float):
return round(obj, 2)
elif isinstance(obj, int):
return obj # int不处理，保留原样（如有需要可加round）
elif isinstance(obj, list):
return [round_floats(x) for x in obj]
elif isinstance(obj, dict):
return {k: round_floats(v) for k, v in obj.items()}
else:
return obj
result = round_floats(result)
# =====================================================
return result
except Exception as e:
logger.error(f"获取概念板块综合分析失败: {e}")
return {"success": False, "message": f"获取概念板块综合分析失败: {e}"}