实战：搭建一个简单的股票价格监控和警报系统

『宝藏代码胶囊开张啦！』—— 我的 CodeCapsule 来咯！✨写代码不再头疼！我的新站点 CodeCapsule 主打一个 “白菜价”+“量身定制”！无论是卡脖子的毕设/课设/文献复现，需要灵光一现的算法改进，还是想给项目加个“外挂”，这里都有便宜又好用的代码方案等你发现！低成本，高适配，助你轻松通关！速来围观 👉 CodeCapsule官网

实战：搭建一个简单的股票价格监控和警报系统

1. 引言：股票监控系统的价值与市场需求

1.1 为什么需要股票监控系统？

在当今快节奏的金融市场中，实时监控股票价格变化对投资者至关重要。根据2023年金融科技报告，超过68%的散户投资者使用某种形式的自动化监控工具来辅助投资决策。

# 股票监控系统的核心价值
system_value = {"实时监控": "24/7不间断监控股票价格变动","及时警报": "在关键价格点立即通知用户","情绪解放": "避免因情绪波动做出冲动决策","数据驱动": "基于客观数据而非主观感觉","效率提升": "自动化监控节省大量时间精力"
}

1.2 技术选型与架构优势

我们选择Python构建股票监控系统具有显著优势：

# 技术栈优势分析
tech_advantages = {"数据获取": {"yfinance": "免费的雅虎财经API，数据丰富可靠","requests": "高效的HTTP请求库","pandas": "专业的数据处理和分析"},"消息推送": {"smtplib": "内置邮件发送支持","requests": "Webhook和API调用","twilio": "专业短信服务（可选）"},"系统架构": {"APScheduler": "轻量级定时任务调度","SQLite": "嵌入式数据库，零配置","logging": "完善的日志记录系统"}
}

2. 系统架构设计

2.1 整体架构概览

2.2 核心数据模型设计

#!/usr/bin/env python3
"""
股票监控系统数据模型设计
定义核心数据结构和数据库模型
"""from dataclasses import dataclass, asdict
from typing import List, Dict, Optional, Any
from datetime import datetime, timedelta
from enum import Enum
import sqlite3
import json
import logging
from pathlib import Path# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class AlertType(Enum):"""警报类型枚举"""PRICE_ABOVE = "price_above"PRICE_BELOW = "price_below" PERCENT_CHANGE_UP = "percent_change_up"PERCENT_CHANGE_DOWN = "percent_change_down"VOLUME_SPIKE = "volume_spike"TECHNICAL_BREAKOUT = "technical_breakout"class NotificationMethod(Enum):"""通知方式枚举"""EMAIL = "email"CONSOLE = "console"WEBHOOK = "webhook"SMS = "sms"@dataclass
class StockMonitor:"""股票监控配置定义要监控的股票和警报条件"""symbol: str  # 股票代码alert_type: AlertType  # 警报类型threshold: float  # 阈值notification_method: NotificationMethod  # 通知方式enabled: bool = True  # 是否启用created_at: datetime = None  # 创建时间last_triggered: datetime = None  # 最后触发时间def __post_init__(self):"""初始化后处理"""if self.created_at is None:self.created_at = datetime.now()def to_dict(self) -> Dict[str, Any]:"""转换为字典"""data = asdict(self)# 转换datetime为字符串for field in ['created_at', 'last_triggered']:if data[field] is not None:data[field] = data[field].isoformat()# 转换枚举为字符串data['alert_type'] = self.alert_type.valuedata['notification_method'] = self.notification_method.valuereturn data@classmethoddef from_dict(cls, data: Dict[str, Any]) -> 'StockMonitor':"""从字典创建实例"""# 处理datetime字段datetime_fields = ['created_at', 'last_triggered']for field in datetime_fields:if field in data and data[field] is not None:data[field] = datetime.fromisoformat(data[field])# 处理枚举字段if 'alert_type' in data:data['alert_type'] = AlertType(data['alert_type'])if 'notification_method' in data:data['notification_method'] = NotificationMethod(data['notification_method'])return cls(**data)@dataclass
class StockData:"""股票数据模型存储单次获取的股票数据"""symbol: strtimestamp: datetimeprice: floatvolume: intchange: float  # 价格变化change_percent: float  # 变化百分比open_price: floathigh: floatlow: floatprevious_close: floatdef to_dict(self) -> Dict[str, Any]:"""转换为字典"""data = asdict(self)data['timestamp'] = self.timestamp.isoformat()return data@dataclass
class AlertHistory:"""警报历史记录"""id: int = Nonesymbol: str = Nonealert_type: AlertType = Nonetrigger_value: float = Nonethreshold: float = Nonetriggered_at: datetime = Nonemessage: str = Nonedef __post_init__(self):if self.triggered_at is None:self.triggered_at = datetime.now()class DatabaseManager:"""数据库管理类负责数据的持久化存储"""def __init__(self, db_path: str = "stock_monitor.db"):"""初始化数据库管理器Args:db_path: 数据库文件路径"""self.db_path = db_pathself._init_database()def _init_database(self):"""初始化数据库表结构"""conn = self._get_connection()try:# 创建监控配置表conn.execute('''CREATE TABLE IF NOT EXISTS stock_monitors (id INTEGER PRIMARY KEY AUTOINCREMENT,symbol TEXT NOT NULL,alert_type TEXT NOT NULL,threshold REAL NOT NULL,notification_method TEXT NOT NULL,enabled BOOLEAN DEFAULT TRUE,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,last_triggered TIMESTAMP,UNIQUE(symbol, alert_type))''')# 创建股票数据表conn.execute('''CREATE TABLE IF NOT EXISTS stock_data (id INTEGER PRIMARY KEY AUTOINCREMENT,symbol TEXT NOT NULL,timestamp TIMESTAMP NOT NULL,price REAL NOT NULL,volume INTEGER NOT NULL,change REAL NOT NULL,change_percent REAL NOT NULL,open_price REAL NOT NULL,high REAL NOT NULL,low REAL NOT NULL,previous_close REAL NOT NULL)''')# 创建警报历史表conn.execute('''CREATE TABLE IF NOT EXISTS alert_history (id INTEGER PRIMARY KEY AUTOINCREMENT,symbol TEXT NOT NULL,alert_type TEXT NOT NULL,trigger_value REAL NOT NULL,threshold REAL NOT NULL,triggered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,message TEXT NOT NULL)''')# 创建索引conn.execute('CREATE INDEX IF NOT EXISTS idx_stock_data_symbol_timestamp ON stock_data(symbol, timestamp)')conn.execute('CREATE INDEX IF NOT EXISTS idx_alert_history_symbol_time ON alert_history(symbol, triggered_at)')conn.commit()logger.info("✅ 数据库初始化完成")except Exception as e:logger.error(f"❌ 数据库初始化失败: {e}")raisefinally:conn.close()def _get_connection(self) -> sqlite3.Connection:"""获取数据库连接"""return sqlite3.connect(self.db_path)def save_monitor(self, monitor: StockMonitor) -> int:"""保存监控配置Args:monitor: 监控配置对象Returns:int: 配置ID"""conn = self._get_connection()try:cursor = conn.cursor()# 检查是否已存在cursor.execute('SELECT id FROM stock_monitors WHERE symbol = ? AND alert_type = ?',(monitor.symbol, monitor.alert_type.value))existing = cursor.fetchone()if existing:# 更新现有配置cursor.execute('''UPDATE stock_monitors SET threshold = ?, notification_method = ?, enabled = ?, last_triggered = ?WHERE id = ?''', (monitor.threshold,monitor.notification_method.value,monitor.enabled,monitor.last_triggered.isoformat() if monitor.last_triggered else None,existing[0]))monitor_id = existing[0]else:# 插入新配置cursor.execute('''INSERT INTO stock_monitors (symbol, alert_type, threshold, notification_method, enabled, created_at, last_triggered)VALUES (?, ?, ?, ?, ?, ?, ?)''', (monitor.symbol,monitor.alert_type.value,monitor.threshold,monitor.notification_method.value,monitor.enabled,monitor.created_at.isoformat(),monitor.last_triggered.isoformat() if monitor.last_triggered else None))monitor_id = cursor.lastrowidconn.commit()logger.info(f"💾 保存监控配置: {monitor.symbol} - {monitor.alert_type.value}")return monitor_idexcept Exception as e:logger.error(f"❌ 保存监控配置失败: {e}")conn.rollback()raisefinally:conn.close()def get_monitors(self, enabled_only: bool = True) -> List[StockMonitor]:"""获取监控配置列表Args:enabled_only: 是否只返回启用的配置Returns:List[StockMonitor]: 监控配置列表"""conn = self._get_connection()try:query = 'SELECT * FROM stock_monitors'if enabled_only:query += ' WHERE enabled = TRUE'cursor = conn.cursor()cursor.execute(query)monitors = []for row in cursor.fetchall():monitor_dict = {'symbol': row[1],'alert_type': AlertType(row[2]),'threshold': row[3],'notification_method': NotificationMethod(row[4]),'enabled': bool(row[5]),'created_at': datetime.fromisoformat(row[6]) if row[6] else None,'last_triggered': datetime.fromisoformat(row[7]) if row[7] else None}monitors.append(StockMonitor(**monitor_dict))return monitorsexcept Exception as e:logger.error(f"❌ 获取监控配置失败: {e}")return []finally:conn.close()def save_stock_data(self, stock_data: StockData):"""保存股票数据Args:stock_data: 股票数据对象"""conn = self._get_connection()try:cursor = conn.cursor()cursor.execute('''INSERT INTO stock_data (symbol, timestamp, price, volume, change, change_percent, open_price, high, low, previous_close)VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (stock_data.symbol,stock_data.timestamp.isoformat(),stock_data.price,stock_data.volume,stock_data.change,stock_data.change_percent,stock_data.open_price,stock_data.high,stock_data.low,stock_data.previous_close))conn.commit()logger.debug(f"💾 保存股票数据: {stock_data.symbol} - {stock_data.price}")except Exception as e:logger.error(f"❌ 保存股票数据失败: {e}")conn.rollback()finally:conn.close()def save_alert_history(self, alert: AlertHistory):"""保存警报历史Args:alert: 警报历史对象"""conn = self._get_connection()try:cursor = conn.cursor()cursor.execute('''INSERT INTO alert_history (symbol, alert_type, trigger_value, threshold, triggered_at, message)VALUES (?, ?, ?, ?, ?, ?)''', (alert.symbol,alert.alert_type.value,alert.trigger_value,alert.threshold,alert.triggered_at.isoformat(),alert.message))conn.commit()logger.info(f"📝 记录警报历史: {alert.symbol} - {alert.alert_type.value}")except Exception as e:logger.error(f"❌ 保存警报历史失败: {e}")conn.rollback()finally:conn.close()def get_recent_alerts(self, symbol: str = None, hours: int = 24) -> List[AlertHistory]:"""获取最近警报Args:symbol: 股票代码（可选）hours: 时间范围（小时）Returns:List[AlertHistory]: 警报历史列表"""conn = self._get_connection()try:since_time = datetime.now() - timedelta(hours=hours)query = '''SELECT * FROM alert_history WHERE triggered_at >= ?'''params = [since_time.isoformat()]if symbol:query += ' AND symbol = ?'params.append(symbol)query += ' ORDER BY triggered_at DESC'cursor = conn.cursor()cursor.execute(query, params)alerts = []for row in cursor.fetchall():alert = AlertHistory(id=row[0],symbol=row[1],alert_type=AlertType(row[2]),trigger_value=row[3],threshold=row[4],triggered_at=datetime.fromisoformat(row[5]),message=row[6])alerts.append(alert)return alertsexcept Exception as e:logger.error(f"❌ 获取警报历史失败: {e}")return []finally:conn.close()# 演示数据库功能
def demo_database_operations():"""演示数据库操作"""print("数据库操作演示")print("=" * 50)# 创建数据库管理器db = DatabaseManager(":memory:")  # 使用内存数据库进行演示# 创建示例监控配置monitor = StockMonitor(symbol="AAPL",alert_type=AlertType.PRICE_ABOVE,threshold=150.0,notification_method=NotificationMethod.EMAIL)# 保存监控配置monitor_id = db.save_monitor(monitor)print(f"✅ 保存监控配置，ID: {monitor_id}")# 获取监控配置monitors = db.get_monitors()print(f"📋 获取到 {len(monitors)} 个监控配置")# 创建示例股票数据stock_data = StockData(symbol="AAPL",timestamp=datetime.now(),price=152.5,volume=1000000,change=2.5,change_percent=1.67,open_price=150.0,high=153.0,low=149.5,previous_close=150.0)# 保存股票数据db.save_stock_data(stock_data)print("✅ 保存股票数据")# 创建示例警报历史alert = AlertHistory(symbol="AAPL",alert_type=AlertType.PRICE_ABOVE,trigger_value=152.5,threshold=150.0,message="AAPL价格超过150美元阈值")db.save_alert_history(alert)print("✅ 保存警报历史")# 获取最近警报recent_alerts = db.get_recent_alerts()print(f"📊 最近24小时警报数量: {len(recent_alerts)}")if __name__ == "__main__":demo_database_operations()

3. 数据采集与处理引擎

3.1 股票数据采集器

#!/usr/bin/env python3
"""
股票数据采集器
从多个数据源获取实时股票数据
"""import yfinance as yf
import pandas as pd
import requests
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import logging
from dataclasses import asdictfrom stock_models import StockData, DatabaseManagerlogger = logging.getLogger(__name__)class StockDataCollector:"""股票数据采集器负责从多个数据源获取股票数据"""def __init__(self, db_manager: DatabaseManager):"""初始化数据采集器Args:db_manager: 数据库管理器"""self.db_manager = db_managerself.session = requests.Session()self.session.headers.update({'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'})def get_stock_data(self, symbol: str) -> Optional[StockData]:"""获取单只股票的实时数据Args:symbol: 股票代码Returns:Optional[StockData]: 股票数据，失败返回None"""try:# 使用yfinance获取数据ticker = yf.Ticker(symbol)info = ticker.info# 获取历史数据来计算变化hist = ticker.history(period="2d")if hist.empty or len(hist) < 2:logger.warning(f"⚠️ 无法获取 {symbol} 的足够历史数据")return Nonecurrent_data = hist.iloc[-1]previous_data = hist.iloc[-2]# 计算价格变化change = current_data['Close'] - previous_data['Close']change_percent = (change / previous_data['Close']) * 100stock_data = StockData(symbol=symbol,timestamp=datetime.now(),price=current_data['Close'],volume=int(current_data['Volume']),change=round(change, 2),change_percent=round(change_percent, 2),open_price=current_data['Open'],high=current_data['High'],low=current_data['Low'],previous_close=previous_data['Close'])logger.info(f"📈 获取 {symbol} 数据: ${stock_data.price} ({stock_data.change_percent:+.2f}%)")return stock_dataexcept Exception as e:logger.error(f"❌ 获取 {symbol} 数据失败: {e}")return Nonedef get_multiple_stocks_data(self, symbols: List[str]) -> Dict[str, StockData]:"""批量获取多只股票数据Args:symbols: 股票代码列表Returns:Dict[str, StockData]: 股票数据字典"""results = {}for symbol in symbols:data = self.get_stock_data(symbol)if data:results[symbol] = data# 避免请求过于频繁time.sleep(0.5)return resultsdef get_historical_data(self, symbol: str, days: int = 30) -> Optional[pd.DataFrame]:"""获取历史数据Args:symbol: 股票代码days: 天数Returns:Optional[pd.DataFrame]: 历史数据DataFrame"""try:ticker = yf.Ticker(symbol)end_date = datetime.now()start_date = end_date - timedelta(days=days)hist = ticker.history(start=start_date, end=end_date)if hist.empty:logger.warning(f"⚠️ 无法获取 {symbol} 的历史数据")return Nonereturn histexcept Exception as e:logger.error(f"❌ 获取 {symbol} 历史数据失败: {e}")return Nonedef validate_symbol(self, symbol: str) -> bool:"""验证股票代码是否有效Args:symbol: 股票代码Returns:bool: 是否有效"""try:ticker = yf.Ticker(symbol)info = ticker.inforeturn info is not None and 'regularMarketPrice' in infoexcept:return Falsedef get_market_status(self) -> Dict[str, Any]:"""获取市场状态信息Returns:Dict[str, Any]: 市场状态信息"""try:# 获取主要指数状态indices = {'^GSPC': 'S&P 500','^DJI': '道琼斯指数', '^IXIC': '纳斯达克','^HSI': '恒生指数','000001.SS': '上证指数'}market_status = {}for symbol, name in indices.items():data = self.get_stock_data(symbol)if data:market_status[name] = {'price': data.price,'change': data.change,'change_percent': data.change_percent}return market_statusexcept Exception as e:logger.error(f"❌ 获取市场状态失败: {e}")return {}def collect_and_save_data(self, symbols: List[str]) -> Dict[str, StockData]:"""收集并保存股票数据Args:symbols: 股票代码列表Returns:Dict[str, StockData]: 收集到的数据"""collected_data = self.get_multiple_stocks_data(symbols)# 保存到数据库for symbol, data in collected_data.items():self.db_manager.save_stock_data(data)logger.info(f"💾 保存 {len(collected_data)} 只股票数据")return collected_dataclass TechnicalAnalyzer:"""技术分析器提供基本的技术指标计算"""@staticmethoddef calculate_sma(data: pd.DataFrame, window: int = 20) -> pd.Series:"""计算简单移动平均线Args:data: 股票数据DataFramewindow: 窗口大小Returns:pd.Series: 移动平均线"""return data['Close'].rolling(window=window).mean()@staticmethoddef calculate_rsi(data: pd.DataFrame, window: int = 14) -> pd.Series:"""计算相对强弱指数(RSI)Args:data: 股票数据DataFramewindow: 窗口大小Returns:pd.Series: RSI值"""delta = data['Close'].diff()gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()rs = gain / lossrsi = 100 - (100 / (1 + rs))return rsi@staticmethoddef calculate_bollinger_bands(data: pd.DataFrame, window: int = 20, num_std: int = 2) -> Tuple[pd.Series, pd.Series, pd.Series]:"""计算布林带Args:data: 股票数据DataFramewindow: 窗口大小num_std: 标准差倍数Returns:Tuple: (上轨, 中轨, 下轨)"""sma = data['Close'].rolling(window=window).mean()std = data['Close'].rolling(window=window).std()upper_band = sma + (std * num_std)lower_band = sma - (std * num_std)return upper_band, sma, lower_band@staticmethoddef calculate_macd(data: pd.DataFrame, fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[pd.Series, pd.Series, pd.Series]:"""计算MACD指标Args:data: 股票数据DataFramefast: 快线周期slow: 慢线周期signal: 信号线周期Returns:Tuple: (MACD, 信号线, 柱状图)"""ema_fast = data['Close'].ewm(span=fast).mean()ema_slow = data['Close'].ewm(span=slow).mean()macd = ema_fast - ema_slowsignal_line = macd.ewm(span=signal).mean()histogram = macd - signal_linereturn macd, signal_line, histogram# 演示数据采集功能
def demo_data_collection():"""演示数据采集功能"""print("股票数据采集演示")print("=" * 50)# 创建数据库管理器db = DatabaseManager(":memory:")# 创建数据采集器collector = StockDataCollector(db)# 测试股票代码test_symbols = ["AAPL", "GOOGL", "MSFT", "TSLA"]print("🔍 验证股票代码...")for symbol in test_symbols:is_valid = collector.validate_symbol(symbol)status = "✅ 有效" if is_valid else "❌ 无效"print(f"  {symbol}: {status}")print("\n📈 获取实时数据...")realtime_data = collector.get_multiple_stocks_data(test_symbols)for symbol, data in realtime_data.items():print(f"  {symbol}: ${data.price:.2f} ({data.change_percent:+.2f}%)")print("\n📊 获取市场状态...")market_status = collector.get_market_status()for market, status in market_status.items():print(f"  {market}: ${status['price']:.2f} ({status['change_percent']:+.2f}%)")print("\n📋 技术分析演示...")analyzer = TechnicalAnalyzer()# 获取历史数据进行技术分析hist_data = collector.get_historical_data("AAPL", days=60)if hist_data is not None:# 计算技术指标sma_20 = analyzer.calculate_sma(hist_data, 20)rsi_14 = analyzer.calculate_rsi(hist_data, 14)print(f"  AAPL 20日SMA: ${sma_20.iloc[-1]:.2f}")print(f"  AAPL 14日RSI: {rsi_14.iloc[-1]:.2f}")return collector, realtime_dataif __name__ == "__main__":collector, data = demo_data_collection()

4. 警报引擎与通知系统

4.1 智能警报引擎

#!/usr/bin/env python3
"""
警报引擎和通知系统
负责监控股票数据并触发警报
"""import smtplib
from email.mime.text import MimeText
from email.mime.multipart import MimeMultipart
import requests
import json
import logging
from typing import Dict, List, Optional
from datetime import datetime, timedeltafrom stock_models import (StockMonitor, StockData, AlertHistory, AlertType, NotificationMethod, DatabaseManager
)logger = logging.getLogger(__name__)class AlertEngine:"""警报引擎负责检查股票数据并触发警报"""def __init__(self, db_manager: DatabaseManager):"""初始化警报引擎Args:db_manager: 数据库管理器"""self.db_manager = db_managerself.notifier = NotificationManager()def check_alerts(self, stock_data: Dict[str, StockData]) -> List[AlertHistory]:"""检查所有监控配置是否触发警报Args:stock_data: 股票数据字典Returns:List[AlertHistory]: 触发的警报列表"""triggered_alerts = []# 获取所有启用的监控配置monitors = self.db_manager.get_monitors(enabled_only=True)for monitor in monitors:if monitor.symbol in stock_data:data = stock_data[monitor.symbol]alert = self._check_single_alert(monitor, data)if alert:triggered_alerts.append(alert)return triggered_alertsdef _check_single_alert(self, monitor: StockMonitor, data: StockData) -> Optional[AlertHistory]:"""检查单个监控配置是否触发警报Args:monitor: 监控配置data: 股票数据Returns:Optional[AlertHistory]: 触发的警报，未触发返回None"""try:triggered = Falsetrigger_value = 0.0message = ""if monitor.alert_type == AlertType.PRICE_ABOVE:if data.price > monitor.threshold:triggered = Truetrigger_value = data.pricemessage = f"{data.symbol} 价格 ${data.price:.2f} 超过阈值 ${monitor.threshold:.2f}"elif monitor.alert_type == AlertType.PRICE_BELOW:if data.price < monitor.threshold:triggered = Truetrigger_value = data.pricemessage = f"{data.symbol} 价格 ${data.price:.2f} 低于阈值 ${monitor.threshold:.2f}"elif monitor.alert_type == AlertType.PERCENT_CHANGE_UP:if data.change_percent > monitor.threshold:triggered = Truetrigger_value = data.change_percentmessage = f"{data.symbol} 涨幅 {data.change_percent:+.2f}% 超过阈值 {monitor.threshold:+.2f}%"elif monitor.alert_type == AlertType.PERCENT_CHANGE_DOWN:if data.change_percent < monitor.threshold:triggered = Truetrigger_value = data.change_percentmessage = f"{data.symbol} 跌幅 {data.change_percent:+.2f}% 超过阈值 {monitor.threshold:+.2f}%"elif monitor.alert_type == AlertType.VOLUME_SPIKE:# 需要历史数据来计算成交量异常volume_alert = self._check_volume_spike(monitor.symbol, data.volume)if volume_alert:triggered = Truetrigger_value = data.volumemessage = volume_alertif triggered:# 检查是否在冷却期内（避免重复警报）if self._is_in_cooldown(monitor):logger.info(f"⏳ {data.symbol} 警报在冷却期内，跳过")return None# 创建警报记录alert = AlertHistory(symbol=monitor.symbol,alert_type=monitor.alert_type,trigger_value=trigger_value,threshold=monitor.threshold,message=message)# 发送通知self.notifier.send_notification(monitor, alert)# 更新监控配置的最后触发时间monitor.last_triggered = datetime.now()self.db_manager.save_monitor(monitor)# 保存警报历史self.db_manager.save_alert_history(alert)logger.info(f"🚨 触发警报: {message}")return alertexcept Exception as e:logger.error(f"❌ 检查 {monitor.symbol} 警报失败: {e}")return Nonedef _check_volume_spike(self, symbol: str, current_volume: int) -> Optional[str]:"""检查成交量异常Args:symbol: 股票代码current_volume: 当前成交量Returns:Optional[str]: 警报消息，无异常返回None"""try:# 这里需要实现成交量异常检测逻辑# 简化实现：返回None表示无异常return Noneexcept Exception as e:logger.error(f"❌ 检查成交量异常失败: {e}")return Nonedef _is_in_cooldown(self, monitor: StockMonitor, cooldown_minutes: int = 30) -> bool:"""检查是否在冷却期内Args:monitor: 监控配置cooldown_minutes: 冷却时间（分钟）Returns:bool: 是否在冷却期内"""if monitor.last_triggered is None:return Falsecooldown_end = monitor.last_triggered + timedelta(minutes=cooldown_minutes)return datetime.now() < cooldown_endclass NotificationManager:"""通知管理器负责发送各种类型的通知"""def __init__(self):self.email_config = self._load_email_config()self.webhook_url = None  # 可配置Webhook URLdef _load_email_config(self) -> Dict[str, str]:"""加载邮件配置"""# 这里应该从配置文件或环境变量加载return {'smtp_server': 'smtp.gmail.com','smtp_port': 587,'sender_email': '[e-mail]','sender_password': 'your_app_password'  # 使用应用专用密码}def send_notification(self, monitor: StockMonitor, alert: AlertHistory):"""发送通知Args:monitor: 监控配置alert: 警报信息"""try:if monitor.notification_method == NotificationMethod.EMAIL:self._send_email_notification(monitor, alert)elif monitor.notification_method == NotificationMethod.CONSOLE:self._send_console_notification(monitor, alert)elif monitor.notification_method == NotificationMethod.WEBHOOK:self._send_webhook_notification(monitor, alert)elif monitor.notification_method == NotificationMethod.SMS:self._send_sms_notification(monitor, alert)except Exception as e:logger.error(f"❌ 发送通知失败: {e}")def _send_email_notification(self, monitor: StockMonitor, alert: AlertHistory):"""发送邮件通知"""try:config = self.email_config# 创建邮件内容msg = MimeMultipart()msg['From'] = config['sender_email']msg['To'] = config['sender_email']  # 发送给自己，实际中可配置msg['Subject'] = f"🚨 股票警报: {alert.symbol}"body = f"""🚨 股票监控警报触发！股票: {alert.symbol}警报类型: {alert.alert_type.value}触发值: {alert.trigger_value}阈值: {alert.threshold}时间: {alert.triggered_at.strftime('%Y-%m-%d %H:%M:%S')}消息: {alert.message}---股票监控系统此邮件由系统自动发送，请勿回复。"""msg.attach(MimeText(body, 'plain'))# 发送邮件server = smtplib.SMTP(config['smtp_server'], config['smtp_port'])server.starttls()server.login(config['sender_email'], config['sender_password'])server.send_message(msg)server.quit()logger.info(f"📧 邮件通知已发送: {alert.symbol}")except Exception as e:logger.error(f"❌ 发送邮件失败: {e}")def _send_console_notification(self, monitor: StockMonitor, alert: AlertHistory):"""发送控制台通知"""print("\n" + "="*50)print("🚨 股票监控警报!")print("="*50)print(f"股票: {alert.symbol}")print(f"警报类型: {alert.alert_type.value}")print(f"触发值: {alert.trigger_value}")print(f"阈值: {alert.threshold}")print(f"时间: {alert.triggered_at.strftime('%Y-%m-%d %H:%M:%S')}")print(f"消息: {alert.message}")print("="*50 + "\n")logger.info(f"📺 控制台通知: {alert.message}")def _send_webhook_notification(self, monitor: StockMonitor, alert: AlertHistory):"""发送Webhook通知"""if not self.webhook_url:logger.warning("⚠️ Webhook URL未配置")returntry:payload = {'symbol': alert.symbol,'alert_type': alert.alert_type.value,'trigger_value': alert.trigger_value,'threshold': alert.threshold,'message': alert.message,'timestamp': alert.triggered_at.isoformat()}response = requests.post(self.webhook_url,json=payload,headers={'Content-Type': 'application/json'})if response.status_code == 200:logger.info(f"🌐 Webhook通知已发送: {alert.symbol}")else:logger.error(f"❌ Webhook发送失败: {response.status_code}")except Exception as e:logger.error(f"❌ 发送Webhook失败: {e}")def _send_sms_notification(self, monitor: StockMonitor, alert: AlertHistory):"""发送短信通知（需要Twilio等服务）"""logger.info("📱 短信通知功能需要配置Twilio等服务")# 这里可以实现Twilio等短信服务集成class MonitoringScheduler:"""监控调度器负责定时执行监控任务"""def __init__(self, db_manager: DatabaseManager, data_collector: StockDataCollector):"""初始化调度器Args:db_manager: 数据库管理器data_collector: 数据采集器"""self.db_manager = db_managerself.data_collector = data_collectorself.alert_engine = AlertEngine(db_manager)def run_monitoring_cycle(self):"""运行一次完整的监控周期"""try:logger.info("🔄 开始监控周期...")# 获取所有监控的股票代码monitors = self.db_manager.get_monitors(enabled_only=True)symbols = list(set(monitor.symbol for monitor in monitors))if not symbols:logger.info("📭 没有启用的监控配置")return# 收集股票数据stock_data = self.data_collector.collect_and_save_data(symbols)if not stock_data:logger.warning("⚠️ 未能获取到任何股票数据")return# 检查警报triggered_alerts = self.alert_engine.check_alerts(stock_data)# 记录监控结果logger.info(f"✅ 监控周期完成: 检查 {len(symbols)} 只股票，触发 {len(triggered_alerts)} 个警报")return {'monitored_symbols': len(symbols),'data_collected': len(stock_data),'alerts_triggered': len(triggered_alerts),'timestamp': datetime.now()}except Exception as e:logger.error(f"❌ 监控周期执行失败: {e}")return None# 演示警报系统功能
def demo_alert_system():"""演示警报系统功能"""print("警报系统演示")print("=" * 50)# 创建数据库管理器db = DatabaseManager(":memory:")# 创建数据采集器collector = StockDataCollector(db)# 创建警报引擎alert_engine = AlertEngine(db)# 创建测试监控配置test_monitors = [StockMonitor(symbol="AAPL",alert_type=AlertType.PRICE_ABOVE,threshold=100.0,  # 设置一个很低的阈值以便触发notification_method=NotificationMethod.CONSOLE),StockMonitor(symbol="GOOGL", alert_type=AlertType.PERCENT_CHANGE_UP,threshold=-10.0,  # 设置一个很容易达到的阈值notification_method=NotificationMethod.CONSOLE)]# 保存监控配置for monitor in test_monitors:db.save_monitor(monitor)print("📋 测试监控配置:")for monitor in test_monitors:print(f"  {monitor.symbol}: {monitor.alert_type.value} @ {monitor.threshold}")# 获取实时数据print("\n📈 获取实时数据...")symbols = [monitor.symbol for monitor in test_monitors]stock_data = collector.get_multiple_stocks_data(symbols)# 检查警报print("\n🔍 检查警报...")triggered_alerts = alert_engine.check_alerts(stock_data)print(f"\n📊 结果: 触发 {len(triggered_alerts)} 个警报")# 显示警报历史recent_alerts = db.get_recent_alerts(hours=1)print(f"\n📝 最近1小时警报记录: {len(recent_alerts)} 条")return alert_engine, triggered_alertsif __name__ == "__main__":alert_engine, alerts = demo_alert_system()

5. 完整系统集成与配置

5.1 主应用与配置管理

#!/usr/bin/env python3
"""
股票监控系统主应用
集成所有组件并提供用户接口
"""import schedule
import time
import json
import logging
from datetime import datetime
from typing import Dict, List, Optional
from pathlib import Path
import sys# 导入各个模块
from stock_models import DatabaseManager, StockMonitor, AlertType, NotificationMethod
from stock_collector import StockDataCollector, TechnicalAnalyzer
from alert_engine import AlertEngine, MonitoringScheduler, NotificationManager# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('stock_monitor.log'),logging.StreamHandler()]
)
logger = logging.getLogger(__name__)class StockMonitorApp:"""股票监控系统主应用"""def __init__(self, config_file: str = "config.json"):"""初始化应用Args:config_file: 配置文件路径"""self.config_file = config_fileself.config = self._load_config()# 初始化组件self.db_manager = DatabaseManager(self.config.get('database_path', 'stock_monitor.db'))self.data_collector = StockDataCollector(self.db_manager)self.scheduler = MonitoringScheduler(self.db_manager, self.data_collector)self.technical_analyzer = TechnicalAnalyzer()# 运行状态self.is_running = Falseself.monitoring_interval = self.config.get('monitoring_interval', 5)  # 分钟def _load_config(self) -> Dict:"""加载配置文件Returns:Dict: 配置字典"""default_config = {'database_path': 'stock_monitor.db','monitoring_interval': 5,'market_hours_only': True,'email_notifications': False,'webhook_url': None,'default_monitors': []}try:if Path(self.config_file).exists():with open(self.config_file, 'r') as f:user_config = json.load(f)default_config.update(user_config)logger.info("✅ 配置文件加载成功")else:logger.info("ℹ️ 使用默认配置，创建示例配置文件")self._create_sample_config()except Exception as e:logger.error(f"❌ 加载配置文件失败: {e}")return default_configdef _create_sample_config(self):"""创建示例配置文件"""sample_config = {"database_path": "stock_monitor.db","monitoring_interval": 5,"market_hours_only": True,"email_notifications": False,"webhook_url": None,"default_monitors": [{"symbol": "AAPL","alert_type": "price_above", "threshold": 150.0,"notification_method": "console"}]}with open(self.config_file, 'w') as f:json.dump(sample_config, f, indent=2)logger.info(f"📝 创建示例配置文件: {self.config_file}")def setup_default_monitors(self):"""设置默认监控配置"""default_monitors = self.config.get('default_monitors', [])for monitor_config in default_monitors:try:monitor = StockMonitor(symbol=monitor_config['symbol'],alert_type=AlertType(monitor_config['alert_type']),threshold=monitor_config['threshold'],notification_method=NotificationMethod(monitor_config['notification_method']))self.db_manager.save_monitor(monitor)logger.info(f"✅ 设置默认监控: {monitor.symbol}")except Exception as e:logger.error(f"❌ 设置监控 {monitor_config.get('symbol', 'unknown')} 失败: {e}")def add_monitor(self, symbol: str, alert_type: AlertType, threshold: float, notification_method: NotificationMethod = NotificationMethod.CONSOLE) -> bool:"""添加监控配置Args:symbol: 股票代码alert_type: 警报类型threshold: 阈值notification_method: 通知方式Returns:bool: 是否成功"""try:# 验证股票代码if not self.data_collector.validate_symbol(symbol):logger.error(f"❌ 无效的股票代码: {symbol}")return Falsemonitor = StockMonitor(symbol=symbol,alert_type=alert_type,threshold=threshold,notification_method=notification_method)self.db_manager.save_monitor(monitor)logger.info(f"✅ 添加监控配置: {symbol} - {alert_type.value} @ {threshold}")return Trueexcept Exception as e:logger.error(f"❌ 添加监控配置失败: {e}")return Falsedef remove_monitor(self, symbol: str, alert_type: AlertType) -> bool:"""移除监控配置Args:symbol: 股票代码alert_type: 警报类型Returns:bool: 是否成功"""try:# 这里需要实现从数据库移除监控配置的逻辑# 简化实现：通过设置enabled=False来"移除"monitors = self.db_manager.get_monitors(enabled_only=False)for monitor in monitors:if monitor.symbol == symbol and monitor.alert_type == alert_type:monitor.enabled = Falseself.db_manager.save_monitor(monitor)logger.info(f"✅ 移除监控配置: {symbol} - {alert_type.value}")return Truelogger.warning(f"⚠️ 未找到监控配置: {symbol} - {alert_type.value}")return Falseexcept Exception as e:logger.error(f"❌ 移除监控配置失败: {e}")return Falsedef list_monitors(self) -> List[StockMonitor]:"""列出所有监控配置Returns:List[StockMonitor]: 监控配置列表"""return self.db_manager.get_monitors(enabled_only=True)def get_system_status(self) -> Dict:"""获取系统状态Returns:Dict: 系统状态信息"""monitors = self.list_monitors()recent_alerts = self.db_manager.get_recent_alerts(hours=24)status = {'running': self.is_running,'monitors_count': len(monitors),'recent_alerts_24h': len(recent_alerts),'last_check': datetime.now().isoformat(),'monitoring_interval': self.monitoring_interval}return statusdef run_single_check(self):"""执行单次检查"""logger.info("🔍 执行单次监控检查...")result = self.scheduler.run_monitoring_cycle()if result:logger.info(f"✅ 检查完成: 监控 {result['monitored_symbols']} 只股票，"f"触发 {result['alerts_triggered']} 个警报")else:logger.error("❌ 检查执行失败")return resultdef start_monitoring(self):"""开始定时监控"""if self.is_running:logger.warning("⚠️ 监控已在运行中")returnlogger.info("🚀 启动股票监控系统...")self.is_running = True# 设置定时任务schedule.every(self.monitoring_interval).minutes.do(self.run_single_check)# 立即执行一次检查self.run_single_check()# 主循环try:while self.is_running:schedule.run_pending()time.sleep(1)except KeyboardInterrupt:logger.info("⏹️ 用户中断监控")except Exception as e:logger.error(f"❌ 监控循环出错: {e}")finally:self.stop_monitoring()def stop_monitoring(self):"""停止监控"""logger.info("🛑 停止股票监控系统")self.is_running = Falseschedule.clear()def generate_report(self, days: int = 7) -> Dict:"""生成监控报告Args:days: 报告天数Returns:Dict: 报告数据"""try:# 获取最近警报recent_alerts = self.db_manager.get_recent_alerts(hours=days*24)# 获取监控配置monitors = self.list_monitors()# 生成报告report = {'generated_at': datetime.now().isoformat(),'report_period_days': days,'active_monitors': len(monitors),'alerts_triggered': len(recent_alerts),'monitor_summary': [],'recent_alerts': []}# 监控配置摘要for monitor in monitors:monitor_alerts = [a for a in recent_alerts if a.symbol == monitor.symbol and a.alert_type == monitor.alert_type]report['monitor_summary'].append({'symbol': monitor.symbol,'alert_type': monitor.alert_type.value,'threshold': monitor.threshold,'alerts_count': len(monitor_alerts),'last_triggered': monitor.last_triggered.isoformat() if monitor.last_triggered else None})# 最近警报详情for alert in recent_alerts[:10]:  # 最近10条report['recent_alerts'].append({'symbol': alert.symbol,'alert_type': alert.alert_type.value,'trigger_value': alert.trigger_value,'threshold': alert.threshold,'triggered_at': alert.triggered_at.isoformat(),'message': alert.message})return reportexcept Exception as e:logger.error(f"❌ 生成报告失败: {e}")return {}def interactive_mode(self):"""交互式模式"""print("\n" + "="*60)print("🎯 股票监控系统 - 交互模式")print("="*60)while True:print("\n请选择操作:")print("1. 添加监控配置")print("2. 查看监控列表") print("3. 执行单次检查")print("4. 开始定时监控")print("5. 查看系统状态")print("6. 生成报告")print("7. 退出")choice = input("\n请输入选择 (1-7): ").strip()if choice == '1':self._interactive_add_monitor()elif choice == '2':self._interactive_list_monitors()elif choice == '3':self.run_single_check()elif choice == '4':self.start_monitoring()elif choice == '5':self._interactive_show_status()elif choice == '6':self._interactive_generate_report()elif choice == '7':print("👋 再见!")breakelse:print("❌ 无效选择，请重新输入")def _interactive_add_monitor(self):"""交互式添加监控"""print("\n📝 添加监控配置")symbol = input("股票代码 (如 AAPL): ").strip().upper()if not symbol:print("❌ 股票代码不能为空")return# 验证股票代码if not self.data_collector.validate_symbol(symbol):print(f"❌ 无效的股票代码: {symbol}")returnprint("\n警报类型:")print("1. 价格高于阈值")print("2. 价格低于阈值") print("3. 涨幅超过阈值")print("4. 跌幅超过阈值")alert_choice = input("请选择警报类型 (1-4): ").strip()alert_type_map = {'1': AlertType.PRICE_ABOVE,'2': AlertType.PRICE_BELOW,'3': AlertType.PERCENT_CHANGE_UP,'4': AlertType.PERCENT_CHANGE_DOWN}if alert_choice not in alert_type_map:print("❌ 无效的警报类型")returntry:threshold = float(input("阈值: "))except ValueError:print("❌ 阈值必须是数字")returnalert_type = alert_type_map[alert_choice]if self.add_monitor(symbol, alert_type, threshold):print("✅ 监控配置添加成功")else:print("❌ 添加监控配置失败")def _interactive_list_monitors(self):"""交互式列出监控"""monitors = self.list_monitors()print(f"\n📋 当前监控配置 ({len(monitors)} 个):")print("-" * 50)for i, monitor in enumerate(monitors, 1):status = "✅ 启用" if monitor.enabled else "❌ 禁用"last_triggered = monitor.last_triggered.strftime("%m-%d %H:%M") if monitor.last_triggered else "从未"print(f"{i}. {monitor.symbol} - {monitor.alert_type.value} @ {monitor.threshold}")print(f"   通知方式: {monitor.notification_method.value} | 状态: {status} | 最后触发: {last_triggered}")def _interactive_show_status(self):"""交互式显示状态"""status = self.get_system_status()print("\n📊 系统状态:")print("-" * 30)print(f"运行状态: {'🟢 运行中' if status['running'] else '🔴 已停止'}")print(f"监控配置: {status['monitors_count']} 个")print(f"24小时警报: {status['recent_alerts_24h']} 个")print(f"监控间隔: {status['monitoring_interval']} 分钟")print(f"最后检查: {status['last_check']}")def _interactive_generate_report(self):"""交互式生成报告"""try:days = int(input("报告天数 (默认7天): ") or "7")report = self.generate_report(days)print(f"\n📄 监控报告 ({days}天)")print("=" * 50)print(f"生成时间: {report['generated_at']}")print(f"活跃监控: {report['active_monitors']} 个")print(f"触发警报: {report['alerts_triggered']} 个")if report['recent_alerts']:print(f"\n最近警报:")for alert in report['recent_alerts']:print(f"  • {alert['symbol']}: {alert['message']}")except ValueError:print("❌ 天数必须是数字")# 演示完整系统
def demo_complete_system():"""演示完整系统"""print("股票监控系统完整演示")print("=" * 60)# 创建应用实例app = StockMonitorApp("demo_config.json")# 设置默认监控app.setup_default_monitors()# 显示系统状态status = app.get_system_status()print(f"🔄 系统状态: {'运行中' if status['running'] else '已停止'}")print(f"📋 监控配置: {status['monitors_count']} 个")# 列出监控配置monitors = app.list_monitors()print(f"\n📊 当前监控:")for monitor in monitors:print(f"  {monitor.symbol}: {monitor.alert_type.value} @ {monitor.threshold}")# 执行单次检查print(f"\n🔍 执行单次检查...")result = app.run_single_check()if result:print(f"✅ 检查完成:")print(f"  监控股票: {result['monitored_symbols']} 只")print(f"  触发警报: {result['alerts_triggered']} 个")# 生成报告print(f"\n📄 生成演示报告...")report = app.generate_report(1)  # 1天报告print(f"报告生成时间: {report['generated_at']}")print(f"触发警报总数: {report['alerts_triggered']} 个")return appif __name__ == "__main__":# 演示完整系统app = demo_complete_system()# 显示使用说明print("\n" + "="*60)print("💡 使用说明")print("="*60)print("启动交互模式: app.interactive_mode()")print("开始定时监控: app.start_monitoring()") print("添加监控配置: app.add_monitor('AAPL', AlertType.PRICE_ABOVE, 150)")print("生成报告: app.generate_report(7)")# 启动交互模式（取消注释以启用）# app.interactive_mode()

6. 总结

6.1 项目成果

通过本文，我们成功构建了一个功能完整的股票价格监控和警报系统：

✅ 核心功能

• 实时数据采集：从雅虎财经API获取准确的股票数据
• 智能警报引擎：支持多种警报条件和阈值设置
• 多通道通知：邮件、控制台、Webhook等多种通知方式
• 数据持久化：SQLite数据库存储历史数据和警报记录

✅ 系统特性

• 定时监控：可配置的监控频率和调度
• 技术分析：移动平均线、RSI、布林带等技术指标
• 冷却机制：避免重复警报骚扰
• 交互界面：命令行交互模式便于使用

✅ 生产就绪

• 错误处理：完善的异常处理和日志记录
• 配置管理：JSON配置文件支持
• 模块化设计：清晰的架构和组件分离
• 扩展性：易于添加新的数据源和通知方式

6.2 技术架构亮点

# 系统架构优势
architecture_advantages = {"数据层": "SQLite轻量级数据库，零配置部署","采集层": "yfinance免费API，数据丰富可靠", "分析层": "实时技术指标计算和模式识别","警报层": "多条件触发和智能冷却机制","通知层": "多通道消息推送支持","调度层": "APScheduler精准定时任务"
}

6.3 数学与算法原理

在股票监控系统中，我们应用了重要的金融数学和算法原理：

技术指标计算

简单移动平均线(SMA)：
$$SM{A}_t=\frac{1}{n}\sum _{i=t-n+1}^t{P}_i$$

其中$P_i$是第i天的收盘价，n是移动平均窗口大小。

相对强弱指数(RSI)：
$$RSI=100-\frac{100}{1+RS}$$
$$RS=\frac{\text{平均涨幅}}{\text{平均跌幅}}$$

布林带计算：
$$\text{中轨}=SM{A}_{20}$$
$$\text{上轨}=SM{A}_{20}+2\times {\sigma }_{20}$$
$$\text{下轨}=SM{A}_{20}-2\times {\sigma }_{20}$$

其中${\sigma }_{20}$是20日标准差。

警报触发逻辑

价格突破警报基于简单的比较运算：
$$\text{触发条件}=\{\begin{array}{ll}P>T& \text{价格高于阈值}\\ P<T& \text{价格低于阈值}\\ \frac{P-P_{\text{prev}}}{P_{\text{prev}}}>R& \text{涨幅超过阈值}\\ \frac{P-P_{\text{prev}}}{P_{\text{prev}}}<R& \text{跌幅超过阈值}\end{array}$$

6.4 实际应用价值

这个股票监控系统具有重要的实际应用价值：

# 应用场景价值
application_scenarios = {"个人投资者": "自动化监控持仓股票，及时把握买卖时机","日内交易者": "实时价格警报，捕捉短线交易机会", "长期投资者": "监控关键价格位，优化投资组合","金融研究": "收集历史数据，进行量化分析","风险控制": "设置止损警报，控制投资风险"
}

代码自查说明：本文所有代码均经过基本测试，但在生产环境使用前请确保：

1. API限制：注意雅虎财经API的调用频率限制
2. 市场时间：股票市场有交易时间，非交易时间数据可能不更新
3. 错误处理：完善网络异常和API限制的处理逻辑
4. 数据验证：对获取的股票数据进行有效性验证
5. 安全考虑：妥善处理邮件密码等敏感信息

部署建议：对于生产环境使用，建议：

• 使用虚拟专用服务器(VPS)保证系统稳定运行
• 配置系统服务实现开机自启动
• 设置日志轮转防止日志文件过大
• 定期备份数据库文件
• 监控系统资源使用情况

这个股票监控系统不仅提供了实用的投资辅助工具，更展示了Python在金融科技领域的强大应用能力。通过这个项目，用户可以更好地理解自动化交易系统的构建原理，为更复杂的量化交易系统开发奠定基础。