AI股票分析师数据库课程设计：MySQL金融数据分析实践-平芜编程栈

AI股票分析师数据库课程设计：MySQL金融数据分析实践

最近在折腾一个挺有意思的开源项目——daily_stock_analysis，一个AI驱动的股票分析系统。用了一段时间后，我发现它背后其实藏着一个挺典型的数据库应用场景。每天自动抓取行情数据、分析股票走势、生成报告，这些操作都离不开数据库的支持。

今天我就想从这个实际项目出发，聊聊怎么用MySQL来搭建一个金融数据分析系统。这其实挺适合作为数据库课程的实践案例，既有实际应用价值，又能学到不少数据库设计的门道。

1. 项目背景与数据库需求

daily_stock_analysis这个项目，本质上是一个数据驱动的AI分析系统。它每天自动运行，从多个数据源获取股票行情、新闻资讯，然后通过AI模型分析，最后生成投资建议推送到你的微信或邮箱。

整个过程涉及的数据流转挺有意思的：

数据采集：从AkShare、Tushare等数据源获取实时行情
数据存储：把获取的数据存起来，方便后续分析
数据处理：AI模型读取历史数据进行分析
结果输出：生成分析报告并推送

这里面，数据库扮演着核心角色。没有好的数据库设计，数据采集、存储、查询都会变得很麻烦。

我刚开始用这个项目时，就遇到了几个实际问题：

历史数据查询慢，每次分析都要重新拉取
不同数据源的数据格式不一致，处理起来费劲
想回溯某个时间点的分析结果，找不到历史记录

这些问题的根源，其实都是数据库设计没做好。下面我就分享一下，怎么用MySQL来解决这些问题。

2. 数据库设计思路

设计金融数据分析的数据库，有几个关键点需要考虑：

数据特点：股票数据是时间序列数据，每天都有新数据产生，历史数据量会越来越大。而且数据维度多，除了价格，还有成交量、技术指标、新闻舆情等。

查询需求：最常见的查询是按股票代码和时间范围查历史数据，还有就是按日期查当天的分析结果。

性能要求：数据写入要快（每天定时采集），查询也要快（用户可能随时查看）。

基于这些考虑，我设计了下面这个数据库结构。

2.1 核心表结构设计

先来看看主要的几张表：

-- 股票基本信息表 CREATE TABLE stocks ( id INT AUTO_INCREMENT PRIMARY KEY, symbol VARCHAR(20) NOT NULL COMMENT '股票代码，如600519', market VARCHAR(10) NOT NULL COMMENT '市场类型：A股、港股、美股', name VARCHAR(100) COMMENT '股票名称', industry VARCHAR(50) COMMENT '所属行业', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY idx_symbol (symbol) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='股票基本信息'; -- 每日行情数据表 CREATE TABLE daily_quotes ( id BIGINT AUTO_INCREMENT PRIMARY KEY, stock_id INT NOT NULL COMMENT '关联stocks.id', trade_date DATE NOT NULL COMMENT '交易日期', open_price DECIMAL(10, 2) COMMENT '开盘价', close_price DECIMAL(10, 2) COMMENT '收盘价', high_price DECIMAL(10, 2) COMMENT '最高价', low_price DECIMAL(10, 2) COMMENT '最低价', volume BIGINT COMMENT '成交量', amount DECIMAL(20, 2) COMMENT '成交额', change_percent DECIMAL(5, 2) COMMENT '涨跌幅', ma5 DECIMAL(10, 2) COMMENT '5日均线', ma10 DECIMAL(10, 2) COMMENT '10日均线', ma20 DECIMAL(10, 2) COMMENT '20日均线', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_stock_date (stock_id, trade_date), INDEX idx_date (trade_date), FOREIGN KEY (stock_id) REFERENCES stocks(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='每日行情数据'; -- 技术指标表 CREATE TABLE technical_indicators ( id BIGINT AUTO_INCREMENT PRIMARY KEY, stock_id INT NOT NULL, trade_date DATE NOT NULL, rsi DECIMAL(5, 2) COMMENT 'RSI指标', macd DECIMAL(10, 2) COMMENT 'MACD值', macd_signal DECIMAL(10, 2) COMMENT 'MACD信号线', macd_histogram DECIMAL(10, 2) COMMENT 'MACD柱状图', boll_upper DECIMAL(10, 2) COMMENT '布林线上轨', boll_middle DECIMAL(10, 2) COMMENT '布林线中轨', boll_lower DECIMAL(10, 2) COMMENT '布林线下轨', kdj_k DECIMAL(5, 2) COMMENT 'KDJ K值', kdj_d DECIMAL(5, 2) COMMENT 'KDJ D值', kdj_j DECIMAL(5, 2) COMMENT 'KDJ J值', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY idx_stock_date (stock_id, trade_date), FOREIGN KEY (stock_id) REFERENCES stocks(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='技术指标数据'; -- AI分析结果表 CREATE TABLE ai_analysis ( id BIGINT AUTO_INCREMENT PRIMARY KEY, stock_id INT NOT NULL, analysis_date DATE NOT NULL COMMENT '分析日期', signal_type ENUM('买入', '观望', '卖出') COMMENT '信号类型', confidence_score DECIMAL(3, 2) COMMENT '置信度分数', buy_price DECIMAL(10, 2) COMMENT '建议买入价', stop_loss_price DECIMAL(10, 2) COMMENT '止损价', target_price DECIMAL(10, 2) COMMENT '目标价', analysis_summary TEXT COMMENT '分析摘要', technical_analysis TEXT COMMENT '技术面分析', sentiment_analysis TEXT COMMENT '舆情分析', risk_factors TEXT COMMENT '风险因素', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_stock_date (stock_id, analysis_date), INDEX idx_date_signal (analysis_date, signal_type), FOREIGN KEY (stock_id) REFERENCES stocks(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='AI分析结果';

2.2 设计考虑要点

这样设计有几个好处：

1. 数据分离：把基础信息、行情数据、技术指标、分析结果分开存储，每张表职责单一，维护起来方便。

2. 索引优化：在经常查询的字段上加了索引，比如(stock_id, trade_date)这个组合索引，查某只股票的历史数据会很快。

3. 外键约束：通过外键确保数据一致性，删除股票时，相关的行情数据和分析结果也会自动删除。

4. 适当冗余：在daily_quotes表里直接存了MA5、MA10、MA20这些常用指标，虽然可以通过计算得到，但存起来查询更快。

实际使用中，我还加了几张辅助表：

-- 新闻舆情表 CREATE TABLE news_sentiment ( id BIGINT AUTO_INCREMENT PRIMARY KEY, stock_id INT NOT NULL, news_date DATE NOT NULL, title VARCHAR(500) COMMENT '新闻标题', source VARCHAR(100) COMMENT '新闻来源', sentiment ENUM('正面', '中性', '负面') COMMENT '情感倾向', relevance_score DECIMAL(3, 2) COMMENT '相关度分数', summary TEXT COMMENT '新闻摘要', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_stock_date (stock_id, news_date), FOREIGN KEY (stock_id) REFERENCES stocks(id) ON DELETE CASCADE ); -- 用户自选股表 CREATE TABLE user_watchlist ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(50) NOT NULL COMMENT '用户标识', stock_id INT NOT NULL, added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, notes VARCHAR(500) COMMENT '备注', UNIQUE KEY idx_user_stock (user_id, stock_id), FOREIGN KEY (stock_id) REFERENCES stocks(id) ON DELETE CASCADE );

3. 数据采集与存储优化

数据库设计好了，接下来就是怎么把数据存进去。daily_stock_analysis项目本身支持多种数据源，我们可以借鉴它的思路。

3.1 数据采集脚本

下面是一个简化的数据采集脚本，每天自动运行：

import akshare as ak import pymysql from datetime import datetime, timedelta import time class StockDataCollector: def __init__(self): # 数据库连接配置 self.db_config = { 'host': 'localhost', 'user': 'stock_user', 'password': 'your_password', 'database': 'stock_analysis', 'charset': 'utf8mb4' } self.conn = pymysql.connect(**self.db_config) def get_stock_list(self): """获取需要采集的股票列表""" cursor = self.conn.cursor() cursor.execute("SELECT id, symbol FROM stocks WHERE is_active = 1") return cursor.fetchall() def collect_daily_data(self, stock_symbol, market='A'): """采集单只股票的日线数据""" try: # 使用akshare获取数据 if market == 'A': df = ak.stock_zh_a_hist(symbol=stock_symbol, period="daily") elif market == 'HK': df = ak.stock_hk_hist(symbol=stock_symbol) elif market == 'US': df = ak.stock_us_hist(symbol=stock_symbol) else: return None # 数据清洗和转换 processed_data = [] for _, row in df.iterrows(): data = { 'trade_date': row['日期'], 'open': float(row['开盘']), 'close': float(row['收盘']), 'high': float(row['最高']), 'low': float(row['最低']), 'volume': int(row['成交量']), 'amount': float(row['成交额']) if '成交额' in row else 0 } processed_data.append(data) return processed_data except Exception as e: print(f"采集{stock_symbol}数据失败: {e}") return None def calculate_technical_indicators(self, stock_data): """计算技术指标""" if len(stock_data) < 20: return None # 计算移动平均线 closes = [d['close'] for d in stock_data] indicators = { 'ma5': sum(closes[-5:]) / 5, 'ma10': sum(closes[-10:]) / 10, 'ma20': sum(closes[-20:]) / 20, } # 这里可以添加更多指标计算逻辑 # 比如RSI、MACD、布林带等 return indicators def save_to_database(self, stock_id, daily_data, indicators): """保存数据到数据库""" cursor = self.conn.cursor() try: # 保存日线数据 for data in daily_data: sql = """ INSERT INTO daily_quotes (stock_id, trade_date, open_price, close_price, high_price, low_price, volume, amount, ma5, ma10, ma20) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE open_price = VALUES(open_price), close_price = VALUES(close_price), high_price = VALUES(high_price), low_price = VALUES(low_price), volume = VALUES(volume), amount = VALUES(amount), ma5 = VALUES(ma5), ma10 = VALUES(ma10), ma20 = VALUES(ma20) """ cursor.execute(sql, ( stock_id, data['trade_date'], data['open'], data['close'], data['high'], data['low'], data['volume'], data['amount'], indicators['ma5'], indicators['ma10'], indicators['ma20'] )) self.conn.commit() print(f"股票{stock_id}数据保存成功") except Exception as e: self.conn.rollback() print(f"保存数据失败: {e}") def run_daily_collection(self): """每日数据采集任务""" stocks = self.get_stock_list() for stock_id, symbol in stocks: print(f"正在采集 {symbol} 的数据...") # 采集数据 daily_data = self.collect_daily_data(symbol) if not daily_data: continue # 计算技术指标 indicators = self.calculate_technical_indicators(daily_data) if not indicators: continue # 保存到数据库 self.save_to_database(stock_id, daily_data[-1:], indicators) # 避免请求过快 time.sleep(1) cursor.close() self.conn.close() if __name__ == "__main__": collector = StockDataCollector() collector.run_daily_collection()

3.2 存储优化技巧

数据量大了之后，存储和查询都会变慢。这里有几个优化建议：

1. 分区表：对于daily_quotes这种时间序列数据，可以按时间分区：

-- 按月分区，查询某个月的数据时更快 ALTER TABLE daily_quotes PARTITION BY RANGE (YEAR(trade_date) * 100 + MONTH(trade_date)) ( PARTITION p202401 VALUES LESS THAN (202402), PARTITION p202402 VALUES LESS THAN (202403), PARTITION p202403 VALUES LESS THAN (202404), PARTITION p_future VALUES LESS THAN MAXVALUE );

2. 归档历史数据：早期的数据查询频率低，可以移到归档表：

-- 创建归档表 CREATE TABLE daily_quotes_archive LIKE daily_quotes; -- 移动一年前的数据到归档表 INSERT INTO daily_quotes_archive SELECT * FROM daily_quotes WHERE trade_date < DATE_SUB(CURDATE(), INTERVAL 1 YEAR); -- 删除原表中的归档数据 DELETE FROM daily_quotes WHERE trade_date < DATE_SUB(CURDATE(), INTERVAL 1 YEAR);

3. 使用中间表：对于复杂的统计查询，可以用中间表预计算结果：

-- 创建周统计中间表 CREATE TABLE weekly_stats ( stock_id INT NOT NULL, week_start DATE NOT NULL, week_close DECIMAL(10, 2), week_volume BIGINT, week_change DECIMAL(5, 2), PRIMARY KEY (stock_id, week_start) ); -- 每周更新一次 INSERT INTO weekly_stats SELECT stock_id, DATE_SUB(trade_date, INTERVAL WEEKDAY(trade_date) DAY) as week_start, AVG(close_price) as week_close, SUM(volume) as week_volume, (MAX(close_price) - MIN(close_price)) / MIN(close_price) * 100 as week_change FROM daily_quotes WHERE trade_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) GROUP BY stock_id, week_start ON DUPLICATE KEY UPDATE week_close = VALUES(week_close), week_volume = VALUES(week_volume), week_change = VALUES(week_change);

4. SQL分析查询实战

数据库里有了数据，最重要的就是怎么查、怎么分析。下面我分享几个在实际项目中常用的查询。

4.1 基础查询示例

查询某只股票最近30天的数据：

SELECT trade_date, open_price, close_price, high_price, low_price, volume, ma5, ma10, ma20 FROM daily_quotes dq JOIN stocks s ON dq.stock_id = s.id WHERE s.symbol = '600519' AND trade_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) ORDER BY trade_date DESC;

查询技术指标出现买入信号的股票：

SELECT s.symbol, s.name, dq.trade_date, dq.close_price, ti.rsi, ti.macd, aa.signal_type, aa.confidence_score FROM stocks s JOIN daily_quotes dq ON s.id = dq.stock_id JOIN technical_indicators ti ON s.id = ti.stock_id AND dq.trade_date = ti.trade_date LEFT JOIN ai_analysis aa ON s.id = aa.stock_id AND aa.analysis_date = CURDATE() WHERE dq.trade_date = CURDATE() AND ti.rsi < 30 -- RSI超卖 AND ti.macd > 0 -- MACD金叉 AND aa.signal_type = '买入' ORDER BY aa.confidence_score DESC;

4.2 复杂分析查询

计算股票的相关性分析：

-- 计算两只股票的价格相关性 WITH stock_a AS ( SELECT trade_date, close_price as price_a FROM daily_quotes dq JOIN stocks s ON dq.stock_id = s.id WHERE s.symbol = '600519' AND trade_date >= '2024-01-01' ), stock_b AS ( SELECT trade_date, close_price as price_b FROM daily_quotes dq JOIN stocks s ON dq.stock_id = s.id WHERE s.symbol = '000858' AND trade_date >= '2024-01-01' ) SELECT CORR(sa.price_a, sb.price_b) as correlation, COUNT(*) as sample_count FROM stock_a sa JOIN stock_b sb ON sa.trade_date = sb.trade_date;

分析板块轮动：

-- 分析不同行业的涨跌幅排名 SELECT s.industry, COUNT(DISTINCT s.id) as stock_count, AVG(dq.change_percent) as avg_change, SUM(CASE WHEN dq.change_percent > 0 THEN 1 ELSE 0 END) as up_count, SUM(CASE WHEN dq.change_percent < 0 THEN 1 ELSE 0 END) as down_count FROM stocks s JOIN daily_quotes dq ON s.id = dq.stock_id WHERE dq.trade_date = CURDATE() AND s.industry IS NOT NULL GROUP BY s.industry HAVING stock_count >= 5 -- 只统计至少有5只股票的行业 ORDER BY avg_change DESC;

4.3 性能优化查询

使用窗口函数分析趋势：

-- 分析股票的连续上涨天数 SELECT symbol, trade_date, close_price, change_percent, up_streak FROM ( SELECT s.symbol, dq.trade_date, dq.close_price, dq.change_percent, SUM(CASE WHEN dq.change_percent <= 0 THEN 1 ELSE 0 END) OVER (PARTITION BY s.id ORDER BY dq.trade_date) as reset_flag, ROW_NUMBER() OVER (PARTITION BY s.id ORDER BY dq.trade_date) as rn FROM stocks s JOIN daily_quotes dq ON s.id = dq.stock_id WHERE s.symbol = '600519' AND dq.trade_date >= DATE_SUB(CURDATE(), INTERVAL 60 DAY) ) t CROSS JOIN LATERAL ( SELECT COUNT(*) as up_streak FROM daily_quotes dq2 JOIN stocks s2 ON dq2.stock_id = s2.id WHERE s2.symbol = t.symbol AND dq2.trade_date <= t.trade_date AND dq2.trade_date > DATE_SUB(t.trade_date, INTERVAL 10 DAY) AND dq2.change_percent > 0 ) streak WHERE trade_date = CURDATE();

5. 与AI分析系统集成

数据库设计好了，查询也优化了，最后一步就是怎么和daily_stock_analysis这样的AI系统集成。

5.1 数据供给接口

AI分析需要数据，我们可以提供专门的查询接口：

from flask import Flask, jsonify import pymysql from datetime import datetime, timedelta app = Flask(__name__) def get_db_connection(): return pymysql.connect( host='localhost', user='stock_user', password='your_password', database='stock_analysis', charset='utf8mb4' ) @app.route('/api/stock/<symbol>/history') def get_stock_history(symbol): """获取股票历史数据供AI分析""" conn = get_db_connection() cursor = conn.cursor(pymysql.cursors.DictCursor) # 获取最近60天的数据 cursor.execute(""" SELECT dq.trade_date, dq.open_price, dq.close_price, dq.high_price, dq.low_price, dq.volume, dq.ma5, dq.ma10, dq.ma20, ti.rsi, ti.macd, ns.sentiment, ns.relevance_score FROM stocks s JOIN daily_quotes dq ON s.id = dq.stock_id LEFT JOIN technical_indicators ti ON s.id = ti.stock_id AND dq.trade_date = ti.trade_date LEFT JOIN news_sentiment ns ON s.id = ns.stock_id AND dq.trade_date = ns.news_date WHERE s.symbol = %s AND dq.trade_date >= DATE_SUB(CURDATE(), INTERVAL 60 DAY) ORDER BY dq.trade_date """, (symbol,)) data = cursor.fetchall() cursor.close() conn.close() return jsonify({ 'symbol': symbol, 'history': data, 'last_updated': datetime.now().isoformat() }) @app.route('/api/analysis/save', methods=['POST']) def save_analysis_result(): """保存AI分析结果""" import json from flask import request data = request.json conn = get_db_connection() cursor = conn.cursor() try: sql = """ INSERT INTO ai_analysis (stock_id, analysis_date, signal_type, confidence_score, buy_price, stop_loss_price, target_price, analysis_summary, technical_analysis, sentiment_analysis, risk_factors) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE signal_type = VALUES(signal_type), confidence_score = VALUES(confidence_score), buy_price = VALUES(buy_price), stop_loss_price = VALUES(stop_loss_price), target_price = VALUES(target_price), analysis_summary = VALUES(analysis_summary), technical_analysis = VALUES(technical_analysis), sentiment_analysis = VALUES(sentiment_analysis), risk_factors = VALUES(risk_factors) """ cursor.execute(sql, ( data['stock_id'], data['analysis_date'], data['signal_type'], data['confidence_score'], data['buy_price'], data['stop_loss_price'], data['target_price'], data['analysis_summary'], data['technical_analysis'], data['sentiment_analysis'], data['risk_factors'] )) conn.commit() return jsonify({'status': 'success', 'id': cursor.lastrowid}) except Exception as e: conn.rollback() return jsonify({'status': 'error', 'message': str(e)}), 500 finally: cursor.close() conn.close() if __name__ == '__main__': app.run(debug=True, port=5000)

5.2 定时任务集成

把数据采集和AI分析做成定时任务：

import schedule import time from datetime import datetime def daily_data_pipeline(): """每日数据流水线""" print(f"{datetime.now()} 开始每日数据采集...") # 1. 采集数据 collector = StockDataCollector() collector.run_daily_collection() # 2. 触发AI分析 trigger_ai_analysis() # 3. 生成报告 generate_daily_report() print(f"{datetime.now()} 每日任务完成") def trigger_ai_analysis(): """触发AI分析""" # 这里可以调用daily_stock_analysis的分析功能 # 或者使用本地的大模型进行分析 print("触发AI分析...") # 示例：分析自选股列表 conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT s.id, s.symbol, s.name FROM stocks s JOIN user_watchlist uw ON s.id = uw.stock_id WHERE uw.user_id = 'default' """) stocks = cursor.fetchall() for stock_id, symbol, name in stocks: # 调用AI分析接口 analysis_result = call_ai_analysis(symbol) # 保存分析结果 save_analysis_to_db(stock_id, analysis_result) cursor.close() conn.close() def call_ai_analysis(symbol): """调用AI分析（这里可以集成daily_stock_analysis）""" # 实际项目中，这里会调用真正的AI分析服务 # 为了示例，返回模拟数据 return { 'signal_type': '买入' if hash(symbol) % 3 == 0 else '观望', 'confidence_score': 0.85, 'buy_price': 100.0, 'stop_loss_price': 95.0, 'target_price': 110.0, 'analysis_summary': f"{symbol}技术面良好，建议关注", 'technical_analysis': "MA5>MA10>MA20多头排列", 'sentiment_analysis': "舆情正面", 'risk_factors': "市场波动风险" } # 设置定时任务 schedule.every().day.at("18:00").do(daily_data_pipeline) # 每天18点运行 print("定时任务已启动，等待执行...") while True: schedule.run_pending() time.sleep(60)