#!/usr/bin/env python3 """ 股票数据采集服务 定时采集实时行情数据并存入数据库 数据源: 腾讯财经 (qt.gtimg.cn) """ import os import sys import time import logging import schedule import psycopg2 from psycopg2.extras import execute_values from datetime import datetime # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler('stock_data_service.log') ] ) logger = logging.getLogger(__name__) # 数据库配置 DB_CONFIG = { 'host': os.environ.get('DB_HOST', 'localhost'), 'port': int(os.environ.get('DB_PORT', 5432)), 'database': os.environ.get('DB_NAME', 'stock_app'), 'user': os.environ.get('DB_USER', 'postgres'), 'password': os.environ.get('DB_PASSWORD', '') } def get_db(): """获取数据库连接""" try: return psycopg2.connect(**DB_CONFIG) except Exception as e: logger.error(f"数据库连接失败: {e}") return None def log_update(data_type, status, records_count=0, error_message=None, started_at=None): """记录更新日志""" conn = get_db() if not conn: return try: cur = conn.cursor() cur.execute(""" INSERT INTO data_update_log (data_type, status, records_count, error_message, started_at) VALUES (%s, %s, %s, %s, %s) """, (data_type, status, records_count, error_message, started_at)) conn.commit() except Exception as e: logger.error(f"记录日志失败: {e}") finally: conn.close() def _to_tencent_code(code): """将纯数字股票代码转为腾讯格式 (sh/sz/bj前缀)""" if code.startswith('6'): return f'sh{code}' elif code.startswith('0') or code.startswith('3'): return f'sz{code}' elif code.startswith('8') or code.startswith('4'): return f'bj{code}' else: return f'sz{code}' def _fetch_realtime_from_tencent(stock_codes): """ 从腾讯财经API批量获取实时行情（最佳数据源，腾讯云极快） 腾讯API字段(88个)关键映射: [1]=名称 [2]=代码 [3]=现价 [4]=昨收 [5]=开盘 [6]=成交量(手) [31]=涨跌额 [32]=涨跌% [33]=最高 [34]=最低 [37]=成交额(万) [39]=市盈率 [45]=总市值(亿) [46]=市净率 """ import requests import math def safe_float(val, default=0): try: if val is None or val == '' or val == ' ': return default f = float(val) return default if math.isnan(f) else f except: return default logger.info(f" 使用腾讯财经数据源 ({len(stock_codes)} 只股票)...") tencent_codes = [_to_tencent_code(c) for c in stock_codes] records = [] batch_size = 80 errors = 0 for i in range(0, len(tencent_codes), batch_size): batch = tencent_codes[i:i+batch_size] url = f"http://qt.gtimg.cn/q={','.join(batch)}" try: r = requests.get(url, timeout=15, headers={'Referer': 'https://finance.qq.com'}) if r.status_code != 200: errors += 1 continue lines = r.text.strip().split(';') for line in lines: if '\"' not in line: continue data = line.split('\"')[1] fields = data.split('~') if len(fields) < 40 or not fields[3]: continue code = fields[2] price = safe_float(fields[3]) if price <= 0: continue # 成交量: 腾讯API返回的是手(1手=100股) volume_hands = safe_float(fields[6]) volume = int(volume_hands * 100) # 成交额: 万元 -> 元 amount = safe_float(fields[37]) * 10000 # 总市值: 亿元 -> 元 total_market_cap_yi = safe_float(fields[45]) if len(fields) > 45 else 0 total_market_cap = total_market_cap_yi * 100000000 if total_market_cap_yi > 0 else None records.append(( code, # 代码 fields[1], # 名称 price, # 现价 safe_float(fields[32]), # 涨跌% safe_float(fields[31]), # 涨跌额 volume, # 成交量(股) amount, # 成交额(元) safe_float(fields[33]), # 最高 safe_float(fields[34]), # 最低 safe_float(fields[5]), # 开盘 safe_float(fields[4]), # 昨收 safe_float(fields[39]) if len(fields) > 39 and fields[39].strip() else None, # PE safe_float(fields[46]) if len(fields) > 46 and fields[46].strip() else None, # PB total_market_cap, # 总市值 )) except Exception as e: errors += 1 if errors <= 3: logger.warning(f" 腾讯API批次 {i//batch_size+1} 失败: {e}") import time time.sleep(0.1) # 控制请求频率 if errors > 0: logger.warning(f" 腾讯API共 {errors} 个批次失败") return records if records else None, 'tencent' def update_realtime_prices(): """更新实时价格（全市场A股）- 使用腾讯财经数据源""" started_at = datetime.now() logger.info("开始更新实时价格...") conn = get_db() if not conn: log_update('realtime_price', 'failed', 0, '数据库连接失败', started_at) return try: # 先从DB获取已有的股票代码列表 cur = conn.cursor() cur.execute("SELECT code FROM stock_realtime_price") existing_codes = [r[0] for r in cur.fetchall()] records = None source = None # 腾讯财经数据源 if existing_codes: try: result = _fetch_realtime_from_tencent(existing_codes) if result and result[0]: records, source = result except Exception as e: logger.warning(f" 腾讯数据源失败: {e}") if not records: log_update('realtime_price', 'failed', 0, '所有数据源均失败', started_at) return logger.info(f" 数据源={source}, 获取 {len(records)} 条记录") # 批量插入/更新 cur = conn.cursor() execute_values(cur, """ INSERT INTO stock_realtime_price (code, name, price, change_pct, change_amount, volume, amount, high, low, open, prev_close, pe, pb, total_market_cap, updated_at) VALUES %s ON CONFLICT (code) DO UPDATE SET name = EXCLUDED.name, price = EXCLUDED.price, change_pct = EXCLUDED.change_pct, change_amount = EXCLUDED.change_amount, volume = EXCLUDED.volume, amount = EXCLUDED.amount, high = EXCLUDED.high, low = EXCLUDED.low, open = EXCLUDED.open, prev_close = EXCLUDED.prev_close, pe = COALESCE(EXCLUDED.pe, stock_realtime_price.pe), pb = COALESCE(EXCLUDED.pb, stock_realtime_price.pb), total_market_cap = COALESCE(EXCLUDED.total_market_cap, stock_realtime_price.total_market_cap), updated_at = NOW() """, records, template="(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())") conn.commit() logger.info(f"实时价格更新完成({source}): {len(records)} 条") log_update('realtime_price', 'success', len(records), f'source={source}', started_at) except Exception as e: logger.error(f"更新实时价格失败: {e}") log_update('realtime_price', 'failed', 0, str(e), started_at) finally: conn.close() def _calc_fund_flow_from_5min(conn, target_date=None): """ 从5分钟K线数据计算资金流向（替代东方财富API） 算法: 1. 每根5分钟K线根据 close vs open 判断方向(买入/卖出) 2. 根据成交额(amount)分类: - 超大单: amount >= 100万 - 大单: 20万 <= amount < 100万 - 中单: 4万 <= amount < 20万 - 小单: amount < 4万 3. 主力 = 超大单 + 大单 4. 聚合每只股票的各类净流入 """ from datetime import date as date_cls if target_date is None: target_date = date_cls.today() cur = conn.cursor() # 获取当天所有5分钟K线数据 cur.execute(""" SELECT code, open, close, volume, amount FROM stock_kline_5min WHERE dt::date = %s AND amount > 0 ORDER BY code, dt """, (target_date,)) rows = cur.fetchall() if not rows: return {} # 按股票聚合 from collections import defaultdict stock_flows = defaultdict(lambda: { 'super_buy': 0, 'super_sell': 0, 'big_buy': 0, 'big_sell': 0, 'mid_buy': 0, 'mid_sell': 0, 'small_buy': 0, 'small_sell': 0, 'total_amount': 0 }) for code, open_p, close_p, volume, amount in rows: if not amount or float(amount) <= 0: continue amt = float(amount) sf = stock_flows[code] sf['total_amount'] += amt # 方向: close > open 视为买入, close < open 视为卖出, 相等则各半 is_buy = float(close_p) >= float(open_p) if close_p and open_p else True # 分类 if amt >= 1000000: # 超大单 >= 100万 cat = 'super' elif amt >= 200000: # 大单 >= 20万 cat = 'big' elif amt >= 40000: # 中单 >= 4万 cat = 'mid' else: # 小单 cat = 'small' if is_buy: sf[f'{cat}_buy'] += amt else: sf[f'{cat}_sell'] += amt # 计算各类净流入和占比 results = {} for code, sf in stock_flows.items(): total = sf['total_amount'] if total <= 0: continue super_net = sf['super_buy'] - sf['super_sell'] big_net = sf['big_buy'] - sf['big_sell'] mid_net = sf['mid_buy'] - sf['mid_sell'] small_net = sf['small_buy'] - sf['small_sell'] main_net = super_net + big_net # 主力 = 超大单 + 大单 results[code] = { 'main_net_inflow': round(main_net, 2), 'main_net_inflow_pct': round(main_net / total * 100, 4) if total > 0 else 0, 'super_net_inflow': round(super_net, 2), 'super_net_inflow_pct': round(super_net / total * 100, 4) if total > 0 else 0, 'big_net_inflow': round(big_net, 2), 'big_net_inflow_pct': round(big_net / total * 100, 4) if total > 0 else 0, 'mid_net_inflow': round(mid_net, 2), 'mid_net_inflow_pct': round(mid_net / total * 100, 4) if total > 0 else 0, 'small_net_inflow': round(small_net, 2), 'small_net_inflow_pct': round(small_net / total * 100, 4) if total > 0 else 0, } return results def update_fund_flow_today(): """更新今日资金流向 — 从5分钟K线数据自行计算""" started_at = datetime.now() logger.info("开始更新今日资金流向(从5分钟K线数据计算)...") conn = get_db() if not conn: log_update('fund_flow_today', 'failed', 0, '数据库连接失败', started_at) return try: flows = _calc_fund_flow_from_5min(conn) if not flows: logger.info("今日资金流向: 无5分钟K线数据，跳过") log_update('fund_flow_today', 'skipped', 0, '无5分钟K线数据', started_at) return # 获取实时价格和名称 cur = conn.cursor() codes = list(flows.keys()) cur.execute(""" SELECT code, name, price, change_pct FROM stock_realtime_price WHERE code = ANY(%s) """, (codes,)) price_map = {} for row in cur.fetchall(): price_map[row[0]] = {'name': row[1], 'price': float(row[2] or 0), 'change_pct': float(row[3] or 0)} # 批量写入 records = [] for code, f in flows.items(): info = price_map.get(code, {}) records.append(( code, info.get('name', ''), f['main_net_inflow'], f['main_net_inflow_pct'], f['super_net_inflow'], f['super_net_inflow_pct'], f['big_net_inflow'], f['big_net_inflow_pct'], f['mid_net_inflow'], f['mid_net_inflow_pct'], f['small_net_inflow'], f['small_net_inflow_pct'], info.get('price', 0), info.get('change_pct', 0), )) execute_values(cur, """ INSERT INTO stock_fund_flow_today (code, name, main_net_inflow, main_net_inflow_pct, super_net_inflow, super_net_inflow_pct, big_net_inflow, big_net_inflow_pct, mid_net_inflow, mid_net_inflow_pct, small_net_inflow, small_net_inflow_pct, price, change_pct, updated_at) VALUES %s ON CONFLICT (code) DO UPDATE SET name = EXCLUDED.name, main_net_inflow = EXCLUDED.main_net_inflow, main_net_inflow_pct = EXCLUDED.main_net_inflow_pct, super_net_inflow = EXCLUDED.super_net_inflow, super_net_inflow_pct = EXCLUDED.super_net_inflow_pct, big_net_inflow = EXCLUDED.big_net_inflow, big_net_inflow_pct = EXCLUDED.big_net_inflow_pct, mid_net_inflow = EXCLUDED.mid_net_inflow, mid_net_inflow_pct = EXCLUDED.mid_net_inflow_pct, small_net_inflow = EXCLUDED.small_net_inflow, small_net_inflow_pct = EXCLUDED.small_net_inflow_pct, price = EXCLUDED.price, change_pct = EXCLUDED.change_pct, updated_at = NOW() """, records, template="(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())") conn.commit() logger.info(f"今日资金流向更新完成: {len(records)} 条 (来源: 5分钟K线计算)") log_update('fund_flow_today', 'success', len(records), '来源: 5分钟K线计算', started_at) except Exception as e: logger.error(f"更新今日资金流向失败: {e}") log_update('fund_flow_today', 'failed', 0, str(e), started_at) finally: conn.close() def update_fund_flow_history(stock_codes=None): """更新历史资金流向 — 从5分钟K线数据自行计算 遍历有5分钟K线但尚未写入fund_flow_history的日期，补算资金流向 """ started_at = datetime.now() logger.info("开始更新历史资金流向(从5分钟K线数据计算)...") conn = get_db() if not conn: log_update('fund_flow_history', 'failed', 0, '数据库连接失败', started_at) return try: cur = conn.cursor() # 查找有5分钟K线数据但尚未计算资金流向的日期 cur.execute(""" SELECT DISTINCT dt::date as d FROM stock_kline_5min WHERE dt::date NOT IN ( SELECT DISTINCT trade_date FROM stock_fund_flow_history ) AND dt::date < CURRENT_DATE ORDER BY d DESC LIMIT 30 """) missing_dates = [row[0] for row in cur.fetchall()] if not missing_dates: logger.info("历史资金流向: 无需补算") log_update('fund_flow_history', 'success', 0, '无需补算', started_at) return logger.info(f"需补算 {len(missing_dates)} 天的历史资金流向") total_records = 0 for d in missing_dates: flows = _calc_fund_flow_from_5min(conn, target_date=d) if not flows: continue # 获取当天收盘价和涨跌幅 cur.execute(""" SELECT code, close, change_pct FROM stock_kline_daily WHERE trade_date = %s AND code = ANY(%s) """, (d, list(flows.keys()))) price_map = {} for row in cur.fetchall(): price_map[row[0]] = {'close': float(row[1] or 0), 'change_pct': float(row[2] or 0)} records = [] for code, f in flows.items(): info = price_map.get(code, {}) records.append(( code, d, info.get('close', 0), info.get('change_pct', 0), f['main_net_inflow'], f['main_net_inflow_pct'], f['super_net_inflow'], f['super_net_inflow_pct'], f['big_net_inflow'], f['big_net_inflow_pct'], f['mid_net_inflow'], f['mid_net_inflow_pct'], f['small_net_inflow'], f['small_net_inflow_pct'], )) if records: execute_values(cur, """ INSERT INTO stock_fund_flow_history (code, trade_date, close_price, change_pct, main_net_inflow, main_net_inflow_pct, super_net_inflow, super_net_inflow_pct, big_net_inflow, big_net_inflow_pct, mid_net_inflow, mid_net_inflow_pct, small_net_inflow, small_net_inflow_pct, updated_at) VALUES %s ON CONFLICT (code, trade_date) DO UPDATE SET close_price = EXCLUDED.close_price, change_pct = EXCLUDED.change_pct, main_net_inflow = EXCLUDED.main_net_inflow, main_net_inflow_pct = EXCLUDED.main_net_inflow_pct, super_net_inflow = EXCLUDED.super_net_inflow, super_net_inflow_pct = EXCLUDED.super_net_inflow_pct, big_net_inflow = EXCLUDED.big_net_inflow, big_net_inflow_pct = EXCLUDED.big_net_inflow_pct, mid_net_inflow = EXCLUDED.mid_net_inflow, mid_net_inflow_pct = EXCLUDED.mid_net_inflow_pct, small_net_inflow = EXCLUDED.small_net_inflow, small_net_inflow_pct = EXCLUDED.small_net_inflow_pct, updated_at = NOW() """, records, template="(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())") conn.commit() total_records += len(records) logger.info(f" {d}: {len(records)} 只股票") logger.info(f"历史资金流向补算完成: {total_records} 条记录，{len(missing_dates)} 天") log_update('fund_flow_history', 'success', total_records, f'补算{len(missing_dates)}天, 来源: 5分钟K线计算', started_at) except Exception as e: logger.error(f"更新历史资金流向失败: {e}") log_update('fund_flow_history', 'failed', 0, str(e), started_at) finally: conn.close() def is_trading_time(): """检查当前是否为交易时间""" now = datetime.now() # 周一到周五 if now.weekday() >= 5: return False # 9:15-11:35, 12:55-15:05 hour_min = now.hour * 100 + now.minute return (915 <= hour_min <= 1135) or (1255 <= hour_min <= 1505) def run_scheduled_tasks(): """运行定时任务""" logger.info("股票数据采集服务启动...") # 交易时间每5分钟更新实时价格（腾讯财经数据源） schedule.every(5).minutes.do(lambda: update_realtime_prices() if is_trading_time() else None) # 每天18:05更新今日资金流向（从5分钟K线计算，需在5分钟K线采集17:30后） schedule.every().day.at("18:05").do(update_fund_flow_today) # 每天18:15补算历史资金流向（从5分钟K线计算） schedule.every().day.at("18:15").do(update_fund_flow_history) # 立即执行一次 logger.info("首次执行数据更新...") update_realtime_prices() # 主循环 while True: schedule.run_pending() time.sleep(60) def main(): """主函数""" if len(sys.argv) > 1: cmd = sys.argv[1] if cmd == 'realtime': update_realtime_prices() elif cmd == 'fund_today': update_fund_flow_today() elif cmd == 'fund_history': stock_codes = sys.argv[2:] if len(sys.argv) > 2 else None update_fund_flow_history(stock_codes) elif cmd == 'daemon': run_scheduled_tasks() else: print(f"未知命令: {cmd}") print("用法: python stock_data_service.py [realtime|fund_today|fund_history|daemon]") else: # 默认执行一次实时价格更新 update_realtime_prices() if __name__ == '__main__': main()