""" 市场数据 API 路由 """ from flask import Blueprint, request, jsonify import pandas as pd from datetime import datetime, timedelta from services.stock_service import get_stock_fund_flow, load_cached_data from services.stock_algorithms import get_kline_data as algo_get_kline_data from db import get_db, put_db bp = Blueprint('market', __name__, url_prefix='/api') # ============ 数据库查询API（高速版） ============ @bp.route('/db/realtime_price/', methods=['GET']) def db_realtime_price(stock_code): """从数据库获取实时价格（毫秒级响应）""" conn = get_db() if not conn: return jsonify({'success': False, 'error': '数据库连接失败'}), 500 try: from psycopg2.extras import RealDictCursor cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT code, name, price, change_pct, change_amount, volume, amount, high, low, open, prev_close, pe, pb, total_market_cap, updated_at::text FROM stock_realtime_price WHERE code = %s """, (stock_code,)) row = cur.fetchone() if not row: return jsonify({'success': False, 'error': '未找到数据'}), 404 return jsonify({ 'success': True, 'data': dict(row) }) finally: put_db(conn) @bp.route('/db/realtime_prices', methods=['POST']) def db_realtime_prices(): """批量获取实时价格""" data = request.get_json() codes = data.get('codes', []) if not codes: return jsonify({'success': True, 'data': []}) conn = get_db() if not conn: return jsonify({'success': False, 'error': '数据库连接失败'}), 500 try: from psycopg2.extras import RealDictCursor cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT code, name, price, change_pct, pe, pb, total_market_cap, updated_at::text FROM stock_realtime_price WHERE code = ANY(%s) """, (codes,)) rows = cur.fetchall() return jsonify({ 'success': True, 'data': [dict(row) for row in rows] }) finally: put_db(conn) @bp.route('/db/fund_flow_today/', methods=['GET']) def db_fund_flow_today(stock_code): """从数据库获取今日资金流向""" conn = get_db() if not conn: return jsonify({'success': False, 'error': '数据库连接失败'}), 500 try: from psycopg2.extras import RealDictCursor cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT code, name, main_net_inflow, main_net_inflow_pct, super_net_inflow, super_net_inflow_pct, big_net_inflow, big_net_inflow_pct, price, change_pct, updated_at::text FROM stock_fund_flow_today WHERE code = %s """, (stock_code,)) row = cur.fetchone() if not row: return jsonify({'success': False, 'error': '未找到数据'}), 404 return jsonify({ 'success': True, 'data': dict(row) }) finally: put_db(conn) @bp.route('/db/fund_flow_today_batch', methods=['POST']) def db_fund_flow_today_batch(): """批量获取今日资金流向""" data = request.get_json() codes = data.get('codes', []) if not codes: return jsonify({'success': True, 'data': []}) conn = get_db() if not conn: return jsonify({'success': False, 'error': '数据库连接失败'}), 500 try: from psycopg2.extras import RealDictCursor cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT code, name, main_net_inflow, main_net_inflow_pct, super_net_inflow, super_net_inflow_pct, price, change_pct, updated_at::text FROM stock_fund_flow_today WHERE code = ANY(%s) """, (codes,)) rows = cur.fetchall() return jsonify({ 'success': True, 'data': [dict(row) for row in rows] }) finally: put_db(conn) @bp.route('/db/data_status', methods=['GET']) def db_data_status(): """获取数据更新状态""" conn = get_db() if not conn: return jsonify({'success': False, 'error': '数据库连接失败'}), 500 try: from psycopg2.extras import RealDictCursor cur = conn.cursor(cursor_factory=RealDictCursor) # 获取各表数据统计 cur.execute("SELECT COUNT(*) as count, MAX(updated_at)::text as last_update FROM stock_realtime_price") price_stats = cur.fetchone() cur.execute("SELECT COUNT(*) as count, MAX(updated_at)::text as last_update FROM stock_fund_flow_today") flow_stats = cur.fetchone() cur.execute(""" SELECT data_type, status, records_count, finished_at::text FROM data_update_log ORDER BY finished_at DESC LIMIT 5 """) logs = cur.fetchall() return jsonify({ 'success': True, 'realtime_price': dict(price_stats) if price_stats else {}, 'fund_flow_today': dict(flow_stats) if flow_stats else {}, 'recent_logs': [dict(log) for log in logs] }) finally: put_db(conn) # ============ 原有API（兼容） ============ @bp.route('/hot_stocks', methods=['GET']) def hot_stocks(): """人气榜 — 已删除（东方财富API不可用，无替代源）""" return jsonify({'success': False, 'error': '人气榜功能已停用', 'data': [], 'total': 0}), 410 @bp.route('/kline/', methods=['GET']) def get_kline(stock_code): """获取K线数据 — 使用统一算法模块 services.stock_algorithms""" try: period = request.args.get('period', 'daily') days_map = { 'weekly': 7, 'monthly': 30, 'quarterly': 90, 'yearly': 365 } days = days_map.get(period, 30) # 使用统一K线获取（含5种数据源自动回退） df = algo_get_kline_data(stock_code, days=days, use_local_db=True) if df is None or df.empty: return jsonify({'success': True, 'data': [], 'stock_code': stock_code, 'period': period}) kline_data = [] for _, row in df.iterrows(): d = row.get('date', '') kline_data.append({ 'date': d.strftime('%Y-%m-%d') if hasattr(d, 'strftime') else str(d), 'open': float(row.get('open', 0)), 'close': float(row.get('close', 0)), 'high': float(row.get('high', 0)), 'low': float(row.get('low', 0)), 'volume': float(row.get('volume', 0)), }) return jsonify({ 'success': True, 'data': kline_data, 'stock_code': stock_code, 'period': period }) except Exception as e: print(f"K线接口异常({stock_code}): {e}") return jsonify({'success': True, 'data': [], 'stock_code': stock_code, 'period': period}) @bp.route('/fundflow/', methods=['GET']) def get_fundflow(stock_code): """获取近N天资金流向（失败时返回空数据）""" try: days = request.args.get('days', 3, type=int) try: cached_df, stock_name, _ = load_cached_data(stock_code) except Exception as e: print(f"加载缓存数据失败({stock_code}): {e}") cached_df, stock_name = None, None if cached_df is None or cached_df.empty: try: end_date = datetime.now().strftime('%Y-%m-%d') start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d') cached_df, stock_name, error = get_stock_fund_flow(stock_code, start_date, end_date) except Exception as e: print(f"获取资金流向失败({stock_code}): {e}") return jsonify({'success': True, 'stock_code': stock_code, 'stock_name': '', 'data': []}) if cached_df is None or cached_df.empty: return jsonify({'success': True, 'stock_code': stock_code, 'stock_name': stock_name or '', 'data': []}) cached_df = cached_df.sort_values('日期', ascending=False) recent = cached_df.head(days) flow_data = [] for _, row in recent.iterrows(): flow_data.append({ 'date': row['日期'].strftime('%Y-%m-%d') if hasattr(row['日期'], 'strftime') else str(row['日期']), 'price': float(row['收盘价']) if pd.notna(row['收盘价']) else 0, 'change': float(row['涨跌幅']) if pd.notna(row['涨跌幅']) else 0, 'super_ratio': float(row['超大单净流入-净占比']) if pd.notna(row['超大单净流入-净占比']) else 0, 'main_ratio': float(row['主力净流入-净占比']) if pd.notna(row['主力净流入-净占比']) else 0, }) return jsonify({ 'success': True, 'stock_code': stock_code, 'stock_name': stock_name or '', 'data': flow_data }) except Exception as e: print(f"资金流向接口异常({stock_code}): {e}") return jsonify({'success': True, 'stock_code': stock_code, 'stock_name': '', 'data': []}) @bp.route('/lhb', methods=['GET']) def get_lhb(): """龙虎榜 — 已删除（东方财富API不可用，无替代源）""" return jsonify({'success': False, 'error': '龙虎榜功能已停用', 'data': [], 'total': 0}), 410 @bp.route('/fund_flow_rank', methods=['GET']) def get_fund_flow_rank(): """获取资金流向排行 — 从数据库缓存获取""" try: limit = request.args.get('limit', 50, type=int) # 东方财富API已不可用，从数据库获取缓存数据 from db import get_db as _get_db _conn = _get_db() if _conn: try: _cur = _conn.cursor() _cur.execute(""" SELECT code, name, main_net_inflow, main_net_inflow_pct, price, change_pct FROM stock_fund_flow_today ORDER BY main_net_inflow DESC LIMIT %s """, (limit,)) rows = _cur.fetchall() flow_data = [{'code': r[0], 'name': r[1], 'main_net_inflow': float(r[2] or 0), 'main_pct': float(r[3] or 0), 'price': float(r[4] or 0), 'change_pct': float(r[5] or 0)} for r in rows] return jsonify({'success': True, 'data': flow_data, 'total': len(flow_data), 'source': 'cache'}) finally: put_db(_conn) return jsonify({'success': True, 'data': [], 'total': 0}) except Exception as e: return jsonify({'error': str(e)}), 500 @bp.route('/fundamental/', methods=['GET']) def get_fundamental(stock_code): """获取基本面数据（当日缓存版）+ 近三日资金流向 + 财务指标""" try: from db import db_get_fundamental, db_save_fundamental, get_db from services.stock_service import get_stock_name # 获取近三日资金流向数据 + 技术信号 fund_flow_3days = [] realtime_data = None signal_data = None conn = None try: conn = get_db() if conn: cur = conn.cursor() cur.execute(""" SELECT trade_date, close_price, change_pct, main_net_inflow_pct, super_net_inflow_pct, big_net_inflow_pct FROM stock_fund_flow_history WHERE code = %s ORDER BY trade_date DESC LIMIT 3 """, (stock_code,)) rows = cur.fetchall() for row in rows: fund_flow_3days.append({ 'date': row[0].strftime('%m-%d') if row[0] else '', 'close_price': float(row[1]) if row[1] else 0, 'change_pct': float(row[2]) if row[2] else 0, 'main_pct': float(row[3]) if row[3] else 0, 'super_pct': float(row[4]) if row[4] else 0, 'big_pct': float(row[5]) if row[5] else 0, }) cur.execute(""" SELECT name, price, pe, pb, change_pct, total_market_cap FROM stock_realtime_price WHERE code = %s """, (stock_code,)) rt_row = cur.fetchone() if rt_row: realtime_data = { 'name': rt_row[0], 'price': float(rt_row[1]) if rt_row[1] else None, 'pe': float(rt_row[2]) if rt_row[2] else None, 'pb': float(rt_row[3]) if rt_row[3] else None, 'change_pct': float(rt_row[4]) if rt_row[4] else None, 'total_market_cap': float(rt_row[5]) if rt_row[5] else None, } from datetime import date as date_cls # 优先今天的扫描数据，无则回退到最近可用日期 scan_date = date_cls.today().strftime('%Y-%m-%d') cur.execute(""" SELECT signal_status, indicators, triggered_count FROM stock_signal_scan WHERE code = %s AND scan_date = %s """, (stock_code, scan_date)) sig_row = cur.fetchone() if not sig_row: cur.execute(""" SELECT signal_status, indicators, triggered_count FROM stock_signal_scan WHERE code = %s AND scan_date = (SELECT MAX(scan_date) FROM stock_signal_scan) """, (stock_code,)) sig_row = cur.fetchone() if sig_row: import json as json_mod ss = sig_row[0] if isinstance(sig_row[0], list) else (json_mod.loads(sig_row[0]) if sig_row[0] else []) ind = sig_row[1] if isinstance(sig_row[1], dict) else (json_mod.loads(sig_row[1]) if sig_row[1] else {}) signal_data = { 'signal_status': ss, 'indicators': ind, 'triggered_count': sig_row[2] or 0, } except Exception as e: print(f"获取数据失败: {e}") finally: put_db(conn) # 优先从数据库获取当日缓存 cached = db_get_fundamental(stock_code) if cached: # 优先使用实时价格表中的PE/PB数据 pe_val = realtime_data['pe'] if realtime_data and realtime_data['pe'] else (float(cached['pe']) if cached['pe'] else '') pb_val = realtime_data['pb'] if realtime_data and realtime_data['pb'] else (float(cached['pb']) if cached['pb'] else '') price_val = realtime_data['price'] if realtime_data and realtime_data['price'] else (float(cached['latest_price']) if cached['latest_price'] else '') change_val = realtime_data['change_pct'] if realtime_data and realtime_data['change_pct'] else (float(cached['change_pct']) if cached['change_pct'] else '') market_cap_val = realtime_data['total_market_cap'] if realtime_data and realtime_data['total_market_cap'] else (float(cached['total_market_cap']) if cached['total_market_cap'] else '') return jsonify({ 'success': True, 'data': { 'stock_code': cached['code'], 'stock_name': cached['name'], 'pe_ttm': pe_val, 'pb': pb_val, 'total_market_cap': market_cap_val, 'industry': cached['industry'] or '', 'latest_price': price_val, 'change_pct': change_val, 'roe': float(cached['roe']) if cached.get('roe') else '', 'eps': float(cached['eps']) if cached.get('eps') else '', 'bps': float(cached['bps']) if cached.get('bps') else '', 'revenue_yoy': float(cached['revenue_yoy']) if cached.get('revenue_yoy') else '', 'profit_yoy': float(cached['profit_yoy']) if cached.get('profit_yoy') else '', 'gross_margin': float(cached['gross_margin']) if cached.get('gross_margin') else '', 'net_margin': float(cached['net_margin']) if cached.get('net_margin') else '', 'fund_flow_3days': fund_flow_3days, 'signal_data': signal_data, }, 'source': 'database' }) # 数据库没有基本面缓存，从实时价格表和API获取 result = { 'stock_code': stock_code, 'stock_name': realtime_data['name'] if realtime_data else (get_stock_name(stock_code) or ''), 'pe_ttm': realtime_data['pe'] if realtime_data and realtime_data['pe'] else '', 'pb': realtime_data['pb'] if realtime_data and realtime_data['pb'] else '', 'total_market_cap': realtime_data['total_market_cap'] if realtime_data and realtime_data['total_market_cap'] else '', 'industry': '', 'latest_price': realtime_data['price'] if realtime_data and realtime_data['price'] else '', 'change_pct': realtime_data['change_pct'] if realtime_data and realtime_data['change_pct'] else '', 'roe': '', 'eps': '', 'bps': '', 'revenue_yoy': '', 'profit_yoy': '', 'gross_margin': '', 'net_margin': '', } # 优先使用mairuiapi获取数据 try: from services.mairui_api import get_realtime_price as mairui_realtime, get_financial_indicators, get_company_info # 获取实时价格 rt_result = mairui_realtime(stock_code) if rt_result['success']: rt_data = rt_result['data'] result['latest_price'] = rt_data.get('price', '') result['change_pct'] = rt_data.get('change', '') result['pe_ttm'] = rt_data.get('pe') or result['pe_ttm'] result['pb'] = rt_data.get('pb') or result['pb'] result['total_market_cap'] = rt_data.get('total_market_cap') or result['total_market_cap'] # 获取公司信息 company_result = get_company_info(stock_code) if company_result['success']: company_data = company_result['data'] result['stock_name'] = company_data.get('name') or result['stock_name'] result['industry'] = company_data.get('industry') or result['industry'] # 获取财务指标 fin_result = get_financial_indicators(stock_code) if fin_result['success']: fin_data = fin_result['data'] result['eps'] = fin_data.get('eps') or '' result['bps'] = fin_data.get('bps') or '' result['roe'] = fin_data.get('roe') or '' result['gross_margin'] = fin_data.get('gross_margin') or '' result['net_margin'] = fin_data.get('net_margin') or '' result['revenue_yoy'] = fin_data.get('revenue_yoy') or '' result['profit_yoy'] = fin_data.get('profit_yoy') or '' except Exception as e: print(f"mairuiapi获取基本面失败: {e}") # 备用方案：先试腾讯API，再试akshare try: import requests as _rq _tc = ('sh' if stock_code.startswith('6') else 'sz') + stock_code _rr = _rq.get(f'http://qt.gtimg.cn/q={_tc}', timeout=5, headers={'Referer': 'https://finance.qq.com'}) if _rr.status_code == 200 and '\"' in _rr.text: _ff = _rr.text.split('\"')[1].split('~') if len(_ff) > 46: result['stock_name'] = _ff[1] or result['stock_name'] result['latest_price'] = _ff[3] result['total_market_cap'] = f'{float(_ff[45])*100000000:.0f}' if _ff[45].strip() else '' except Exception as e2: print(f"腾讯财经备用方案也失败: {e2}") # 保存到数据库缓存 try: db_save_fundamental(stock_code, { 'name': result['stock_name'], 'pe': float(result['pe_ttm']) if result['pe_ttm'] else None, 'pb': float(result['pb']) if result['pb'] else None, 'total_market_cap': float(result['total_market_cap']) if result['total_market_cap'] else None, 'industry': result['industry'], 'latest_price': float(result['latest_price']) if result['latest_price'] else None, 'change_pct': float(str(result['change_pct']).replace('%', '')) if result['change_pct'] else None, 'roe': float(result['roe']) if result['roe'] else None, 'eps': float(result['eps']) if result['eps'] else None, 'bps': float(result['bps']) if result['bps'] else None, 'revenue_yoy': float(result['revenue_yoy']) if result['revenue_yoy'] else None, 'profit_yoy': float(result['profit_yoy']) if result['profit_yoy'] else None, 'gross_margin': float(result['gross_margin']) if result['gross_margin'] else None, 'net_margin': float(result['net_margin']) if result['net_margin'] else None, }) except Exception as e: print(f"保存基本面缓存失败: {e}") result['fund_flow_3days'] = fund_flow_3days result['signal_data'] = signal_data return jsonify({'success': True, 'data': result, 'source': 'api'}) except Exception as e: return jsonify({'error': str(e)}), 500 # ============ 市场情绪 & 外部因素 API ============ @bp.route('/market_sentiment', methods=['GET']) def market_sentiment(): """获取市场情绪指标（涨停跌停比、连板高度、换手率中位数、两市成交额）""" try: from services.market_sentiment import calc_market_sentiment result = calc_market_sentiment() return jsonify({'success': True, 'data': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/external_factors', methods=['GET']) def external_factors(): """获取外部因素综合数据（南向资金、美股隔夜、大宗商品、汇率）""" try: from services.external_factors import get_all_external_factors result = get_all_external_factors() return jsonify({'success': True, 'data': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/fund_flow_analysis/', methods=['GET']) def fund_flow_analysis(stock_code): """获取个股资金流向分析（P0：主力资金进出评分和信号）""" try: from services.fund_flow_analyzer import analyze_fund_flow days = request.args.get('days', 5, type=int) result = analyze_fund_flow(stock_code, days=days) return jsonify({'success': True, 'data': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/news_analysis/', methods=['GET']) def news_analysis(stock_code): """获取个股消息面分析（P5公告+P6政策+异动检测）""" try: from services.news_analyzer import analyze_news_factors from services.stock_service import get_stock_name stock_name = get_stock_name(stock_code) or '' result = analyze_news_factors(stock_code, stock_name) return jsonify({'success': True, 'data': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500