告警时间: {message.timestamp.strftime('%Y-%m-%d %H:%M:%S')}
+{message.content.replace(chr(10), '
')}
+ 本邮件由行情数据服务自动发送，请勿回复。 +
+ + + """ + + async def send(self, message: AlertMessage) -> bool: + """发送邮件告警""" + if not self.enabled or not self.to_addrs: + return False + + try: + # 构建邮件 + msg = MIMEMultipart('alternative') + msg['Subject'] = f"[{message.level.upper()}] {message.title}" + msg['From'] = self.from_addr + msg['To'] = ', '.join(self.to_addrs) + + # 添加HTML内容 + html_content = self._build_html_content(message) + msg.attach(MIMEText(html_content, 'html', 'utf-8')) + + # 发送邮件（在executor中执行同步操作） + import asyncio + loop = asyncio.get_event_loop() + + def send_email(): + server = smtplib.SMTP(self.smtp_host, self.smtp_port) + if self.use_tls: + server.starttls() + server.login(self.username, self.password) + server.sendmail(self.from_addr, self.to_addrs, msg.as_string()) + server.quit() + + await loop.run_in_executor(None, send_email) + + info(f"Email alert sent: {message.title}") + return True + + except Exception as e: + error(f"Failed to send email alert: {e}") + return False + + +class WebhookAlertChannel(AlertChannel): + """Webhook告警通道""" + + def __init__( + self, + webhook_url: str, + headers: Optional[Dict[str, str]] = None, + timeout: float = 10.0, + enabled: bool = True + ): + super().__init__("webhook", enabled) + self.webhook_url = webhook_url + self.headers = headers or {"Content-Type": "application/json"} + self.timeout = timeout + + async def send(self, message: AlertMessage) -> bool: + """发送Webhook告警""" + if not self.enabled or not self.webhook_url: + return False + + try: + payload = { + "title": message.title, + "content": message.content, + "level": message.level, + "timestamp": message.timestamp.isoformat(), + "metadata": message.metadata + } + + async with httpx.AsyncClient() as client: + response = await client.post( + self.webhook_url, + json=payload, + headers=self.headers, + timeout=self.timeout + ) + + if response.status_code < 400: + info(f"Webhook alert sent: {message.title}") + return True + else: + error(f"Webhook error: {response.status_code}") + return False + + except Exception as e: + error(f"Failed to send webhook alert: {e}") + return False + + +class AlertManager: + """告警管理器 + + 管理多个告警通道，支持消息路由和批量发送。 + """ + + def __init__(self): + self.channels: Dict[str, AlertChannel] = {} + self.level_routing = { + "info": ["log"], + "warning": ["log", "dingtalk"], + "error": ["log", "dingtalk", "email"], + "critical": ["log", "dingtalk", "email", "webhook"] + } + + def register_channel(self, channel: AlertChannel): + """注册告警通道""" + self.channels[channel.name] = channel + info(f"Alert channel registered: {channel.name}") + + def configure_routing(self, level_routing: Dict[str, List[str]]): + """配置告警路由规则""" + self.level_routing = level_routing + + async def send( + self, + message: AlertMessage, + channels: Optional[List[str]] = None + ) -> Dict[str, bool]: + """发送告警 + + Args: + message: 告警消息 + channels: 指定通道列表，None则根据级别路由 + + Returns: + 各通道发送结果 + """ + # 确定目标通道 + target_channels = channels + if target_channels is None: + target_channels = self.level_routing.get(message.level, ["log"]) + + # 发送到各通道 + results = {} + for channel_name in target_channels: + channel = self.channels.get(channel_name) + if channel: + results[channel_name] = await channel.send(message) + else: + warning(f"Alert channel not found: {channel_name}") + results[channel_name] = False + + return results + + async def send_simple( + self, + title: str, + content: str, + level: str = "warning", + **kwargs + ) -> Dict[str, bool]: + """发送简单告警""" + message = AlertMessage( + title=title, + content=content, + level=level, + metadata=kwargs + ) + return await self.send(message) + + +# 全局告警管理器实例 +_alert_manager: Optional[AlertManager] = None + + +def get_alert_manager() -> AlertManager: + """获取全局告警管理器""" + global _alert_manager + if _alert_manager is None: + _alert_manager = AlertManager() + # 默认注册日志通道 + _alert_manager.register_channel(LogAlertChannel()) + return _alert_manager + + +def init_alert_manager(config: Dict): + """从配置初始化告警管理器""" + global _alert_manager + _alert_manager = AlertManager() + + # 注册日志通道 + _alert_manager.register_channel(LogAlertChannel( + enabled=config.get("log", {}).get("enabled", True) + )) + + # 注册钉钉通道 + dingtalk_config = config.get("dingtalk", {}) + if dingtalk_config.get("enabled"): + _alert_manager.register_channel(DingTalkAlertChannel( + webhook_url=dingtalk_config["webhook_url"], + secret=dingtalk_config.get("secret"), + at_mobiles=dingtalk_config.get("at_mobiles", []), + at_all=dingtalk_config.get("at_all", False), + enabled=True + )) + + # 注册邮件通道 + email_config = config.get("email", {}) + if email_config.get("enabled"): + _alert_manager.register_channel(EmailAlertChannel( + smtp_host=email_config["smtp_host"], + smtp_port=email_config["smtp_port"], + username=email_config["username"], + password=email_config["password"], + from_addr=email_config["from_addr"], + to_addrs=email_config["to_addrs"], + use_tls=email_config.get("use_tls", True), + enabled=True + )) + + # 注册Webhook通道 + webhook_config = config.get("webhook", {}) + if webhook_config.get("enabled"): + _alert_manager.register_channel(WebhookAlertChannel( + webhook_url=webhook_config["webhook_url"], + headers=webhook_config.get("headers"), + timeout=webhook_config.get("timeout", 10.0), + enabled=True + )) + + # 配置路由规则 + if "routing" in config: + _alert_manager.configure_routing(config["routing"]) + + return _alert_manager diff --git a/app/monitor/monitor.py b/app/monitor/monitor.py index 8a0dd57..01d6e87 100644 --- a/app/monitor/monitor.py +++ b/app/monitor/monitor.py @@ -1,6 +1,5 @@ """数据质量监控 - 对应Go的internal/monitor/monitor.go""" import asyncio -from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime, timedelta from typing import List, Optional @@ -11,6 +10,7 @@ from sqlalchemy import text from app.repositories import StockRepository, FuturesRepository from app.models import Frequency from app.core.logger import info, error +from app.monitor.alert_channels import AlertManager, AlertMessage, get_alert_manager @dataclass @@ -37,23 +37,6 @@ class QualityReport: pass_rate: float -class AlertSender(ABC): - """告警发送接口""" - - @abstractmethod - def send_alert(self, title: str, content: str) -> bool: - """发送告警""" - pass - - -class LogAlertSender(AlertSender): - """日志告警发送器""" - - def send_alert(self, title: str, content: str) -> bool: - info(f"[ALERT] {title}: {content}") - return True - - class DataQualityMonitor: """数据质量监控""" @@ -62,12 +45,12 @@ class DataQualityMonitor: db: Session, stock_repo: StockRepository, futures_repo: FuturesRepository, - sender: Optional[AlertSender] = None + alert_manager: Optional[AlertManager] = None ): self.db = db self.stock_repo = stock_repo self.futures_repo = futures_repo - self.sender = sender or LogAlertSender() + self.alert_manager = alert_manager or get_alert_manager() async def daily_check(self, check_date: str): """每日数据质量检查""" @@ -156,11 +139,17 @@ class DataQualityMonitor: result.detail = f"Data missing: expected {expect_count}, actual {actual_count}" # 发送告警 - if self.sender: - self.sender.send_alert( - f"[{asset_type}] Data Missing Alert", - f"Symbol: {symbol}, Date: {check_date}, Expected: {expect_count}, Actual: {actual_count}" - ) + if self.alert_manager: + asyncio.create_task(self.alert_manager.send_simple( + title=f"[{asset_type.upper()}] 数据缺失告警", + content=f"标的: {symbol}, 日期: {check_date}, 期望: {expect_count}条, 实际: {actual_count}条", + level="warning", + asset_type=asset_type, + symbol=symbol, + check_date=check_date, + expect_count=expect_count, + actual_count=actual_count + )) except Exception as e: result.status = "fail" diff --git a/app/repositories/__pycache__/database.cpython-311.pyc b/app/repositories/__pycache__/database.cpython-311.pyc index 08eb722..84f5275 100644 Binary files a/app/repositories/__pycache__/database.cpython-311.pyc and b/app/repositories/__pycache__/database.cpython-311.pyc differ diff --git a/app/repositories/__pycache__/futures_repository.cpython-311.pyc b/app/repositories/__pycache__/futures_repository.cpython-311.pyc index c0ca708..66e9f9e 100644 Binary files a/app/repositories/__pycache__/futures_repository.cpython-311.pyc and b/app/repositories/__pycache__/futures_repository.cpython-311.pyc differ diff --git a/app/repositories/__pycache__/models.cpython-311.pyc b/app/repositories/__pycache__/models.cpython-311.pyc index 953510e..3c501d1 100644 Binary files a/app/repositories/__pycache__/models.cpython-311.pyc and b/app/repositories/__pycache__/models.cpython-311.pyc differ diff --git a/app/repositories/__pycache__/stock_repository.cpython-311.pyc b/app/repositories/__pycache__/stock_repository.cpython-311.pyc index 0754396..a5c1c13 100644 Binary files a/app/repositories/__pycache__/stock_repository.cpython-311.pyc and b/app/repositories/__pycache__/stock_repository.cpython-311.pyc differ diff --git a/app/repositories/futures_repository.py b/app/repositories/futures_repository.py index dbf217a..3395b58 100644 --- a/app/repositories/futures_repository.py +++ b/app/repositories/futures_repository.py @@ -58,10 +58,20 @@ class FuturesRepository: return items def _get_kline_model(self, freq: Frequency): - """根据周期获取K线模型""" + """根据周期获取K线模型 + + 注意: 目前数据库只支持1分钟和日线K线存储。 + 其他周期(5m/15m/30m/60m/1w/1month)默认使用日线表。 + """ mapping = { Frequency.FREQ_1M: FuturesKLine1M, Frequency.FREQ_1D: FuturesKLine1D, + Frequency.FREQ_5M: FuturesKLine1D, # 默认使用日线 + Frequency.FREQ_15M: FuturesKLine1D, + Frequency.FREQ_30M: FuturesKLine1D, + Frequency.FREQ_60M: FuturesKLine1D, + Frequency.FREQ_1W: FuturesKLine1D, + Frequency.FREQ_1MONTH: FuturesKLine1D, } return mapping.get(freq, FuturesKLine1D) diff --git a/app/repositories/models.py b/app/repositories/models.py index c79d26a..410322a 100644 --- a/app/repositories/models.py +++ b/app/repositories/models.py @@ -210,3 +210,20 @@ class DataQualityCheck(Base): actual_count = Column(Integer, nullable=True, comment="实际数量") detail = Column(String(500), nullable=True, comment="详情") created_at = Column(DateTime, default=datetime.now, comment="创建时间") + + +class StockAdjustFactor(Base): + """股票复权系数表""" + __tablename__ = "stock_adjust_factors" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + symbol_id = Column(String(20), nullable=False, index=True, comment="标的代码") + trade_date = Column(String(10), nullable=False, index=True, comment="交易日期 YYYY-MM-DD") + qfq_factor = Column(Numeric(18, 8), nullable=False, default=1.0, comment="前复权系数") + hfq_factor = Column(Numeric(18, 8), nullable=False, default=1.0, comment="后复权系数") + created_at = Column(DateTime, default=datetime.now, comment="创建时间") + updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间") + + __table_args__ = ( + Index("idx_adj_factor_symbol_date", "symbol_id", "trade_date"), + ) diff --git a/app/repositories/stock_repository.py b/app/repositories/stock_repository.py index b3eee4d..03886f4 100644 --- a/app/repositories/stock_repository.py +++ b/app/repositories/stock_repository.py @@ -11,7 +11,7 @@ from app.models import ( ) from app.repositories.models import ( StockSymbol, StockKLine1M, StockKLine5M, StockKLine1D, - StockTradingCalendar + StockTradingCalendar, StockAdjustFactor ) @@ -258,3 +258,85 @@ class StockRepository: self.db.add(new_cal) self.db.commit() + + def get_adjust_factors( + self, + symbol: str, + start_date: str, + end_date: str + ) -> List[dict]: + """获取指定日期范围内的复权系数 + + Args: + symbol: 股票代码 + start_date: 开始日期 (YYYYMMDD) + end_date: 结束日期 (YYYYMMDD) + + Returns: + 复权系数列表，每项包含 trade_date, qfq_factor, hfq_factor + """ + # 转换日期格式 + start_fmt = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:]}" + end_fmt = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:]}" + + results = self.db.query(StockAdjustFactor).filter( + StockAdjustFactor.symbol_id == symbol, + StockAdjustFactor.trade_date >= start_fmt, + StockAdjustFactor.trade_date <= end_fmt + ).order_by(StockAdjustFactor.trade_date.asc()).all() + + return [ + { + "trade_date": r.trade_date, + "qfq_factor": float(r.qfq_factor) if r.qfq_factor else 1.0, + "hfq_factor": float(r.hfq_factor) if r.hfq_factor else 1.0 + } + for r in results + ] + + def save_adjust_factors(self, symbol: str, factors: List[dict]) -> None: + """保存复权系数 + + Args: + symbol: 股票代码 + factors: 复权系数列表，每项包含 trade_date, qfq_factor, hfq_factor + """ + for f in factors: + trade_date = f.get("trade_date") + + existing = self.db.query(StockAdjustFactor).filter( + StockAdjustFactor.symbol_id == symbol, + StockAdjustFactor.trade_date == trade_date + ).first() + + if existing: + existing.qfq_factor = f.get("qfq_factor", 1.0) + existing.hfq_factor = f.get("hfq_factor", 1.0) + else: + new_factor = StockAdjustFactor( + symbol_id=symbol, + trade_date=trade_date, + qfq_factor=f.get("qfq_factor", 1.0), + hfq_factor=f.get("hfq_factor", 1.0) + ) + self.db.add(new_factor) + + self.db.commit() + + def get_latest_adjust_factor(self, symbol: str) -> Optional[dict]: + """获取最新的复权系数 + + Returns: + 包含 qfq_factor 和 hfq_factor 的字典，如果没有则返回None + """ + result = self.db.query(StockAdjustFactor).filter( + StockAdjustFactor.symbol_id == symbol + ).order_by(StockAdjustFactor.trade_date.desc()).first() + + if result: + return { + "trade_date": result.trade_date, + "qfq_factor": float(result.qfq_factor) if result.qfq_factor else 1.0, + "hfq_factor": float(result.hfq_factor) if result.hfq_factor else 1.0 + } + return None diff --git a/app/services/__pycache__/adapter_service.cpython-311.pyc b/app/services/__pycache__/adapter_service.cpython-311.pyc index 4b707c0..95292a2 100644 Binary files a/app/services/__pycache__/adapter_service.cpython-311.pyc and b/app/services/__pycache__/adapter_service.cpython-311.pyc differ diff --git a/app/services/__pycache__/futures_service.cpython-311.pyc b/app/services/__pycache__/futures_service.cpython-311.pyc index c2365a2..ad5eff1 100644 Binary files a/app/services/__pycache__/futures_service.cpython-311.pyc and b/app/services/__pycache__/futures_service.cpython-311.pyc differ diff --git a/app/services/__pycache__/stock_service.cpython-311.pyc b/app/services/__pycache__/stock_service.cpython-311.pyc index 63524a8..b49eea0 100644 Binary files a/app/services/__pycache__/stock_service.cpython-311.pyc and b/app/services/__pycache__/stock_service.cpython-311.pyc differ diff --git a/app/services/adapter_service.py b/app/services/adapter_service.py index 1f2b35e..2d57851 100644 --- a/app/services/adapter_service.py +++ b/app/services/adapter_service.py @@ -203,16 +203,30 @@ class AdapterService: # 从 config.json 获取最新配置（与文件同步） file_config = get_config() print(f"Using file config: {file_config}, adapter name: {name}") - if name == "amazingdata": - # 优先使用 stock 下的 amazingdata 配置 - source_info = file_config.sources.stock.list["amazingdata"] + + # 尝试从配置文件中获取适配器配置 + adapter_config = None + + # 1. 首先检查 stock 配置 + if name in file_config.sources.stock.list: + source_info = file_config.sources.stock.list[name] + adapter_config = dict(source_info.config) if source_info else {} + print(f"Using stock config for {name}: {adapter_config}") + + # 2. 然后检查 futures 配置 + elif name in file_config.sources.futures.list: + source_info = file_config.sources.futures.list[name] adapter_config = dict(source_info.config) if source_info else {} - print(f"Using amazingdata config: {adapter_config}") - # 处理 port 为字符串的情况 - if "port" in adapter_config and isinstance(adapter_config["port"], str): - adapter_config["port"] = int(adapter_config["port"]) if adapter_config["port"].strip() else 8600 + print(f"Using futures config for {name}: {adapter_config}") + + # 3. 使用默认配置 else: adapter_config = self.configs[name].get("config", {}) + print(f"Using default config for {name}: {adapter_config}") + + # 处理 port 为字符串的情况 + if "port" in adapter_config and isinstance(adapter_config["port"], str): + adapter_config["port"] = int(adapter_config["port"]) if adapter_config["port"].strip() else 8600 cfg = {"enabled": self.configs[name].get("enabled", False), "config": adapter_config} diff --git a/app/services/futures_service.py b/app/services/futures_service.py index dd9e218..c549a53 100644 --- a/app/services/futures_service.py +++ b/app/services/futures_service.py @@ -61,10 +61,16 @@ class FuturesService: # 确保适配器已连接 adapter = adapter_service.get_active_adapter("futures") if not adapter: - # 尝试连接 amazingdata + # 从配置获取当前激活的适配器名称 + from app.core.config import get_config + config = get_config() + active_source = config.sources.futures.active + + # 尝试连接配置的适配器 + info(f"Connecting to configured adapter: {active_source}") loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - loop.run_until_complete(adapter_service._connect_adapter("amazingdata")) + loop.run_until_complete(adapter_service._connect_adapter(active_source)) loop.close() adapter = adapter_service.get_active_adapter("futures") @@ -164,7 +170,13 @@ class FuturesService: # 确保适配器已连接 adapter = adapter_service.get_active_adapter("futures") if not adapter: - asyncio.run(adapter_service._connect_adapter("amazingdata")) + # 从配置获取当前激活的适配器名称 + from app.core.config import get_config + config = get_config() + active_source = config.sources.futures.active + + info(f"Connecting to configured adapter: {active_source}") + asyncio.run(adapter_service._connect_adapter(active_source)) adapter = adapter_service.get_active_adapter("futures") if not adapter: diff --git a/app/services/stock_service.py b/app/services/stock_service.py index 2e162ed..9ce7861 100644 --- a/app/services/stock_service.py +++ b/app/services/stock_service.py @@ -71,10 +71,16 @@ class StockService: # 确保适配器已连接 adapter = adapter_service.get_active_adapter("stock") if not adapter: - # 尝试连接 amazingdata + # 从配置获取当前激活的适配器名称 + from app.core.config import get_config + config = get_config() + active_source = config.sources.stock.active + + # 尝试连接配置的适配器 + info(f"Connecting to configured adapter: {active_source}") loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - loop.run_until_complete(adapter_service._connect_adapter("amazingdata")) + loop.run_until_complete(adapter_service._connect_adapter(active_source)) loop.close() adapter = adapter_service.get_active_adapter("stock") @@ -151,13 +157,145 @@ class StockService: def _apply_adjust( self, symbol: str, - items: List, + items: List[KLineItem], adjust_type: AdjustType - ) -> List: - """应用复权计算（TODO: 实现复权逻辑）""" - # 复权计算需要从数据库获取复权系数 - # 这里简化处理，直接返回原始数据 - return items + ) -> List[KLineItem]: + """应用复权计算 + + 复权原理: + - 前复权(qfq): 以最新价格为基准，将历史价格按比例缩小 + - 后复权(hfq): 以历史最早价格为基准，将后续价格按比例放大 + """ + if not items or adjust_type == AdjustType.NONE: + return items + + try: + # 获取日期范围 + start_date = items[0].time.strftime("%Y%m%d") + end_date = items[-1].time.strftime("%Y%m%d") + + # 从数据库获取复权系数 + factors = self.repository.get_adjust_factors(symbol, start_date, end_date) + + # 如果没有复权系数，尝试从适配器获取 + if not factors: + factors = self._fetch_adjust_factors_from_adapter(symbol, start_date, end_date) + if factors: + self.repository.save_adjust_factors(symbol, factors) + + # 将复权系数转换为字典，方便查找 + factor_map = {f["trade_date"]: f for f in factors} + + # 应用复权 + adjusted_items = [] + for item in items: + # 获取交易日期 + trade_date = getattr(item, 'trade_date', None) + if not trade_date and hasattr(item, 'time'): + trade_date = item.time.strftime("%Y-%m-%d") + + factor = factor_map.get(trade_date, {"qfq_factor": 1.0, "hfq_factor": 1.0}) + + # 根据复权类型选择系数 + if adjust_type == AdjustType.QFQ: + adj_factor = factor.get("qfq_factor", 1.0) + else: # HFQ + adj_factor = factor.get("hfq_factor", 1.0) + + # 应用复权系数到价格字段 + adjusted_item = KLineItem( + symbol=item.symbol, + time=item.time, + open=round(item.open * adj_factor, 4), + high=round(item.high * adj_factor, 4), + low=round(item.low * adj_factor, 4), + close=round(item.close * adj_factor, 4), + volume=item.volume, + amount=round(item.amount * adj_factor, 4) if item.amount else item.amount, + trade_date=getattr(item, 'trade_date', None), + is_limit_up=getattr(item, 'is_limit_up', None), + is_limit_down=getattr(item, 'is_limit_down', None), + total_market_cap=getattr(item, 'total_market_cap', None), + float_market_cap=getattr(item, 'float_market_cap', None), + inst_holding_ratio=getattr(item, 'inst_holding_ratio', None), + trading_days=getattr(item, 'trading_days', None), + adj_factor=adj_factor + ) + adjusted_items.append(adjusted_item) + + return adjusted_items + + except Exception as e: + error(f"Failed to apply adjust factor for {symbol}: {e}") + # 出错时返回原始数据 + return items + + def _fetch_adjust_factors_from_adapter( + self, + symbol: str, + start_date: str, + end_date: str + ) -> List[dict]: + """从适配器获取复权系数""" + try: + adapter_service = AdapterService() + adapter = adapter_service.get_active_adapter("stock") + + if not adapter: + error("No active adapter available for fetching adjust factors") + return [] + + # 检查适配器是否支持获取复权因子 + if not hasattr(adapter, 'get_adj_factor'): + return [] + + # 异步获取前复权因子 + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + qfq_df = loop.run_until_complete( + adapter.get_adj_factor([symbol]) + ) + hfq_df = loop.run_until_complete( + adapter.get_backward_factor([symbol]) + ) + finally: + loop.close() + + # 转换DataFrame为列表 + factors = [] + + # 处理日期格式 + for idx in qfq_df.index: + date_obj = idx if hasattr(idx, 'strftime') else datetime.strptime(str(idx), "%Y%m%d") + date_str = date_obj.strftime("%Y-%m-%d") + date_key = date_obj.strftime("%Y%m%d") + + # 只保留指定范围内的数据 + if not (start_date <= date_key <= end_date): + continue + + qfq_factor = float(qfq_df.loc[idx, symbol]) if symbol in qfq_df.columns else 1.0 + hfq_factor = float(hfq_df.loc[idx, symbol]) if symbol in hfq_df.columns else 1.0 + + # 确保复权系数有效 + if qfq_factor <= 0 or qfq_factor != qfq_factor: # 检查NaN + qfq_factor = 1.0 + if hfq_factor <= 0 or hfq_factor != hfq_factor: + hfq_factor = 1.0 + + factors.append({ + "trade_date": date_str, + "qfq_factor": qfq_factor, + "hfq_factor": hfq_factor + }) + + info(f"Fetched {len(factors)} adjust factors from adapter for {symbol}") + return factors + + except Exception as e: + error(f"Failed to fetch adjust factors from adapter: {e}") + return [] def list_symbols(self, req: SymbolListRequest) -> SymbolListData: """查询标的列表""" @@ -196,7 +334,13 @@ class StockService: # 确保适配器已连接 adapter = adapter_service.get_active_adapter("stock") if not adapter: - asyncio.run(adapter_service._connect_adapter("amazingdata")) + # 从配置获取当前激活的适配器名称 + from app.core.config import get_config + config = get_config() + active_source = config.sources.stock.active + + info(f"Connecting to configured adapter: {active_source}") + asyncio.run(adapter_service._connect_adapter(active_source)) adapter = adapter_service.get_active_adapter("stock") if not adapter: @@ -301,7 +445,13 @@ class StockService: # 确保适配器已连接 adapter = adapter_service.get_active_adapter("stock") if not adapter: - asyncio.run(adapter_service._connect_adapter("amazingdata")) + # 从配置获取当前激活的适配器名称 + from app.core.config import get_config + config = get_config() + active_source = config.sources.stock.active + + info(f"Connecting to configured adapter: {active_source}") + asyncio.run(adapter_service._connect_adapter(active_source)) adapter = adapter_service.get_active_adapter("stock") if not adapter: diff --git a/config.json b/config.json index 64e54ea..cf8ee62 100644 --- a/config.json +++ b/config.json @@ -20,32 +20,32 @@ }, "sources": { "stock": { - "active": "custom", + "active": "amazingdata", "list": { - "custom": { + "amazingdata": { "type": "sdk", "config": { - "username": "", - "password": "", - "host": "", - "port": "", - "local_path": "./custom_data_cache/", + "username": "11200008169", + "password": "11200008169@2026", + "host": "140.206.44.234", + "port": "8600", + "local_path": "./amazing_data_cache/", "use_local_cache": "true" } } } }, "futures": { - "active": "custom", + "active": "amazingdata", "list": { - "custom": { + "amazingdata": { "type": "sdk", "config": { "username": "", "password": "", "host": "", - "port": "", - "local_path": "./custom_data_cache/", + "port": "8600", + "local_path": "./amazing_data_cache/", "use_local_cache": "true" } } diff --git a/market_data_service.egg-info/PKG-INFO b/market_data_service.egg-info/PKG-INFO new file mode 100644 index 0000000..bb0d8bd --- /dev/null +++ b/market_data_service.egg-info/PKG-INFO @@ -0,0 +1,269 @@ +Metadata-Version: 2.4 +Name: market-data-service +Version: 1.0.0 +Summary: 统一行情数据服务 - Python实现 +Classifier: Development Status :: 4 - Beta +Classifier: Intended Audience :: Developers +Classifier: Programming Language :: Python :: 3 +Classifier: Programming Language :: Python :: 3.10 +Classifier: Programming Language :: Python :: 3.11 +Classifier: Programming Language :: Python :: 3.12 +Requires-Python: >=3.10 +Description-Content-Type: text/markdown +Requires-Dist: fastapi>=0.115.0 +Requires-Dist: uvicorn[standard]>=0.32.0 +Requires-Dist: python-socketio>=5.12.1 +Requires-Dist: websockets>=14.1 +Requires-Dist: sqlalchemy>=2.0.36 +Requires-Dist: psycopg2-binary>=2.9.10 +Requires-Dist: pandas>=2.2.3 +Requires-Dist: numpy>=2.1.3 +Requires-Dist: numba>=0.61.0 +Requires-Dist: scipy>=1.15.0 +Requires-Dist: pydantic>=2.10.0 +Requires-Dist: pydantic-settings>=2.6.1 +Requires-Dist: python-dotenv>=1.0.1 +Requires-Dist: PyYAML>=6.0.2 +Requires-Dist: httpx>=0.28.0 +Requires-Dist: apscheduler>=3.11.0 +Provides-Extra: dev +Requires-Dist: pytest>=8.3.4; extra == "dev" +Requires-Dist: pytest-asyncio>=0.24.0; extra == "dev" + +# 统一行情数据服务 - Python实现 + +Python版本的统一行情数据服务，所有接口和功能与Go版本保持一致。 + +## 特性 + +- **多周期K线支持**：1m/5m/15m/30m/60m/1d/1w/1month +- **股票复权支持**：前复权(qfq)/后复权(hfq) +- **数据源热切换**：支持Wind、Tushare等多个数据源动态切换 +- **双轨设计**：股票和期货接口独立，数据存储隔离 +- **WebSocket实时订阅**：支持实时行情推送 +- **数据质量监控**：自动检测数据缺失并告警 +- **交易日历**：支持查询股票和期货的交易日历 +- **期货合约查询**：根据品种获取可交易合约列表 + +## 技术栈 + +- **语言**: Python 3.10+ +- **Web框架**: FastAPI +- **WebSocket**: FastAPI原生WebSocket + python-socketio +- **数据库**: PostgreSQL 15+ (SQLAlchemy ORM) +- **数据源**: Tushare (首期支持) + +## 项目结构 + +``` +python_market_data_service/ +├── app/ +│ ├── __init__.py +│ ├── main.py # 主程序入口 +│ ├── api/ # API路由 +│ │ ├── __init__.py +│ │ ├── routes.py # 主要API路由 +│ │ └── admin_routes.py # 管理后台路由 +│ ├── core/ # 核心模块 +│ │ ├── __init__.py +│ │ ├── config.py # 配置管理 +│ │ ├── errors.py # 错误定义 +│ │ └── logger.py # 日志工具 +│ ├── models/ # 数据模型 +│ │ ├── __init__.py +│ │ ├── types.py # 基础类型 +│ │ └── admin_types.py # 管理后台类型 +│ ├── repositories/ # 数据访问层 +│ │ ├── __init__.py +│ │ ├── database.py # 数据库连接 +│ │ ├── models.py # 数据库模型 +│ │ ├── stock_repository.py +│ │ └── futures_repository.py +│ ├── services/ # 业务逻辑层 +│ │ ├── __init__.py +│ │ ├── stock_service.py +│ │ ├── futures_service.py +│ │ ├── admin_service.py +│ │ ├── config_service.py +│ │ ├── adapter_service.py +│ │ └── test_service.py +│ ├── adapters/ # 数据源适配器 +│ │ ├── __init__.py +│ │ ├── base.py # 适配器基类 +│ │ └── tushare_adapter.py +│ └── websocket/ # WebSocket服务 +│ ├── __init__.py +│ └── server.py +├── scripts/ +│ └── sync_data.py # 数据同步工具 +├── tests/ # 测试文件 +├── requirements.txt # 依赖列表 +├── pyproject.toml # 项目配置 +└── README.md # 本文件 +``` + +## 快速开始 + +### 1. 环境准备 + +- Python 3.10+ +- PostgreSQL 15+ +- Tushare Token (从 [Tushare官网](https://tushare.pro) 获取) + +### 2. 安装依赖 + +```bash +# 创建虚拟环境 +python -m venv venv + +# 激活虚拟环境 +# Windows: +venv\Scripts\activate +# Linux/Mac: +source venv/bin/activate + +# 安装依赖 +pip install -r requirements.txt + +# 安装Tushare（需单独安装） +pip install tushare +``` + +### 3. 配置环境变量 + +```bash +# Windows PowerShell +$env:TUSHARE_TOKEN="your_tushare_token" +$env:DATABASE_URL="postgresql://user:password@localhost:5432/marketdata" + +# Linux/Mac +export TUSHARE_TOKEN="your_tushare_token" +export DATABASE_URL="postgresql://user:password@localhost:5432/marketdata" +``` + +### 4. 初始化数据库 + +```bash +# 创建数据库（使用psql或pgAdmin） +createdb marketdata + +# 启动服务时会自动创建表结构 +``` + +### 5. 启动服务 + +```bash +# 开发模式 +python -m app.main + +# 或使用uvicorn +uvicorn app.main:app --reload --port 8080 +``` + +服务将启动在 `http://localhost:8080` + +- API文档: `http://localhost:8080/docs` +- 管理后台: `http://localhost:8080/admin` + +### 6. 同步基础数据 + +```bash +# 同步股票列表 +python scripts/sync_data.py --type stocks + +# 同步期货列表 +python scripts/sync_data.py --type futures + +# 同步交易日历 +python scripts/sync_data.py --type calendar --start 20240101 --end 20241231 + +# 同步K线数据 +python scripts/sync_data.py --type klines --symbol 000001.SZ --start 20240301 --end 20240307 --freq 1d +``` + +## API接口 + +### 股票接口 + +| 接口 | 方法 | 说明 | +|------|------|------| +| `/v1/stock/klines/:symbol` | GET | 查询K线数据 | +| `/v1/stock/symbols` | GET | 查询标的列表 | +| `/v1/stock/klines/batch` | POST | 批量查询K线 | +| `/v1/stock/trading-dates` | GET | 获取交易日历 | + +### 期货接口 + +| 接口 | 方法 | 说明 | +|------|------|------| +| `/v1/futures/klines/:symbol` | GET | 查询K线数据 | +| `/v1/futures/symbols` | GET | 查询标的列表 | +| `/v1/futures/klines/batch` | POST | 批量查询K线 | +| `/v1/futures/continuous/:underlying` | GET | 查询主力连续合约（预留） | +| `/v1/futures/trading-dates` | GET | 获取交易日历 | +| `/v1/futures/contracts` | GET | 获取品种合约列表 | + +### 管理接口 + +| 接口 | 方法 | 说明 | +|------|------|------| +| `/v1/admin/source/status` | GET | 获取数据源状态 | +| `/v1/admin/source/switch` | POST | 切换数据源 | +| `/v1/admin/backfill` | POST | 历史数据补录 | +| `/v1/admin/health` | GET | 健康检查 | + +### 管理后台 + +服务启动后，访问 `http://localhost:8080/admin` 进入管理后台。 + +### WebSocket实时订阅 + +**连接地址**: `ws://localhost:8080/v1/stream` + +**认证**: 连接时在Header中传递 `X-API-Key` + +**客户端消息**: +```json +// 订阅 +{ + "action": "subscribe", + "symbols": ["000001.SZ", "CU2504.SHFE"] +} + +// 取消订阅 +{ + "action": "unsubscribe", + "symbols": ["000001.SZ"] +} +``` + +**服务器消息**: +```json +// 订阅确认 +{ + "type": "ack", + "action": "subscribe", + "symbols": ["000001.SZ", "CU2504.SHFE"], + "ts": "2025-03-07T12:30:00Z" +} + +// 心跳 +{ + "type": "heartbeat", + "ts": "2025-03-07T12:30:30Z" +} +``` + +**限制**: 单连接最大订阅100个标的 + +## 与Go版本的主要区别 + +1. **Web框架**: Gin -> FastAPI +2. **ORM**: 原生SQL -> SQLAlchemy +3. **WebSocket**: Gorilla -> FastAPI原生 +4. **配置**: 文件+环境变量 -> Pydantic Settings +5. **API文档**: 自动生成Swagger/ReDoc + +## License + +MIT diff --git a/market_data_service.egg-info/SOURCES.txt b/market_data_service.egg-info/SOURCES.txt new file mode 100644 index 0000000..ef08a44 --- /dev/null +++ b/market_data_service.egg-info/SOURCES.txt @@ -0,0 +1,43 @@ +README.md +pyproject.toml +app/__init__.py +app/main.py +app/adapters/__init__.py +app/adapters/amazingdata_adapter.py +app/adapters/base.py +app/api/__init__.py +app/api/admin_routes.py +app/api/routes.py +app/core/__init__.py +app/core/config.py +app/core/errors.py +app/core/logger.py +app/core/metrics.py +app/core/rate_limiter.py +app/models/__init__.py +app/models/admin_types.py +app/models/types.py +app/monitor/__init__.py +app/monitor/alert_channels.py +app/monitor/monitor.py +app/repositories/__init__.py +app/repositories/database.py +app/repositories/futures_repository.py +app/repositories/models.py +app/repositories/stock_repository.py +app/services/__init__.py +app/services/adapter_service.py +app/services/admin_service.py +app/services/config_service.py +app/services/futures_service.py +app/services/stock_service.py +app/services/test_service.py +app/websocket/__init__.py +app/websocket/server.py +market_data_service.egg-info/PKG-INFO +market_data_service.egg-info/SOURCES.txt +market_data_service.egg-info/dependency_links.txt +market_data_service.egg-info/requires.txt +market_data_service.egg-info/top_level.txt +tests/test_xysz_adapter.py +tests/test_xysz_integration.py \ No newline at end of file diff --git a/market_data_service.egg-info/dependency_links.txt b/market_data_service.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/market_data_service.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/market_data_service.egg-info/requires.txt b/market_data_service.egg-info/requires.txt new file mode 100644 index 0000000..b5b0438 --- /dev/null +++ b/market_data_service.egg-info/requires.txt @@ -0,0 +1,20 @@ +fastapi>=0.115.0 +uvicorn[standard]>=0.32.0 +python-socketio>=5.12.1 +websockets>=14.1 +sqlalchemy>=2.0.36 +psycopg2-binary>=2.9.10 +pandas>=2.2.3 +numpy>=2.1.3 +numba>=0.61.0 +scipy>=1.15.0 +pydantic>=2.10.0 +pydantic-settings>=2.6.1 +python-dotenv>=1.0.1 +PyYAML>=6.0.2 +httpx>=0.28.0 +apscheduler>=3.11.0 + +[dev] +pytest>=8.3.4 +pytest-asyncio>=0.24.0 diff --git a/market_data_service.egg-info/top_level.txt b/market_data_service.egg-info/top_level.txt new file mode 100644 index 0000000..b80f0bd --- /dev/null +++ b/market_data_service.egg-info/top_level.txt @@ -0,0 +1 @@ +app diff --git a/marketdata.db b/marketdata.db index 457389d..23e36b2 100644 Binary files a/marketdata.db and b/marketdata.db differ diff --git a/marketdata.db.backup b/marketdata.db.backup new file mode 100644 index 0000000..9545876 Binary files /dev/null and b/marketdata.db.backup differ diff --git a/requirements.txt b/requirements.txt index 0a604a2..ac98a7c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -29,6 +29,7 @@ aioredis==2.0.1 # Monitoring apscheduler==3.11.0 +prometheus-client==0.21.0 # Testing pytest==8.3.4 diff --git a/scripts/init_mysql_db.py b/scripts/init_mysql_db.py new file mode 100644 index 0000000..92bab95 --- /dev/null +++ b/scripts/init_mysql_db.py @@ -0,0 +1,53 @@ +"""MySQL数据库初始化脚本 + +创建数据库和表结构 +""" +import sys +sys.path.insert(0, '.') + +from sqlalchemy import create_engine, text +from app.core.config import get_config +from app.repositories.database import Base + +def init_mysql(): + """初始化MySQL数据库""" + config = get_config() + db_config = config.database + + # 连接MySQL服务器（不指定数据库） + server_url = f"mysql+pymysql://{db_config.user}:{db_config.password}@{db_config.host}:{db_config.port}" + + print(f"Connecting to MySQL server: {db_config.host}:{db_config.port}") + engine = create_engine(server_url) + + # 创建数据库 + with engine.connect() as conn: + conn.execute(text(f"CREATE DATABASE IF NOT EXISTS {db_config.database} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")) + print(f"Database '{db_config.database}' created or exists") + + # 连接到新创建的数据库 + db_url = f"{server_url}/{db_config.database}" + db_engine = create_engine(db_url) + + # 创建所有表 + print("Creating tables...") + Base.metadata.create_all(bind=db_engine) + print("Tables created successfully!") + + # 显示创建的表 + with db_engine.connect() as conn: + result = conn.execute(text("SHOW TABLES")) + tables = [row[0] for row in result] + print(f"\nTables in database '{db_config.database}':") + for table in tables: + print(f" - {table}") + +if __name__ == "__main__": + try: + init_mysql() + print("\nMySQL database initialization completed!") + except Exception as e: + print(f"Error: {e}") + import traceback + traceback.print_exc() + sys.exit(1)