Stock MCP Server

"""
简化的增量数据同步器
基于最新日期的简单增量同步策略
"""

import pandas as pd
from typing import Optional, Dict, Any, List
from datetime import datetime, timedelta
import logging
import time

from .tushare_fetcher import TushareMacroFetcher
from ..storage.manager import StorageManager
from ..storage.base import INDICATOR_TIME_FIELD, INDICATOR_FREQUENCY

logger = logging.getLogger(__name__)


class IncrementalSyncEngine:
    """简化的增量同步引擎"""

    def __init__(self):
        """初始化增量同步引擎"""
        self.fetcher = TushareMacroFetcher()
        self.storage_manager = StorageManager()

        # 每个指标的起始时间配置
        self.start_periods = {
            "gdp": "2000Q1",  # GDP从2000年开始
            "cpi": "200001",  # CPI从2000年开始
            "ppi": "200001",  # PPI从2000年开始
            "pmi": "200501",  # PMI从2005年开始有数据
            "money_supply": "200001",  # 货币供应量从2000年开始
            "social_financing": "200201",  # 社融从2002年开始
            "lpr": "20190820",  # LPR从改革开始
        }

        # 数据发布滞后配置
        self.publish_delays = {
            "gdp": {"type": "quarter", "delay": 1},  # 滞后1个季度
            "cpi": {"type": "month", "delay": 1},  # 滞后1个月
            "ppi": {"type": "month", "delay": 1},  # 滞后1个月
            "pmi": {"type": "month", "delay": 1},  # 滞后1个月
            "money_supply": {"type": "month", "delay": 1},  # 滞后1个月
            "social_financing": {"type": "month", "delay": 1},  # 滞后1个月
            "lpr": {"type": "day", "delay": 0},  # 当天发布
        }

        # 每个指标的最小更新间隔（避免过于频繁的同步）
        self.min_sync_interval = {
            "gdp": timedelta(days=30),  # GDP月更新一次
            "cpi": timedelta(days=7),  # CPI周更新一次
            "ppi": timedelta(days=7),  # PPI周更新一次
            "pmi": timedelta(days=7),  # PMI周更新一次
            "money_supply": timedelta(days=7),  # 货币供应量周更新一次
            "social_financing": timedelta(days=7),  # 社融周更新一次
            "lpr": timedelta(days=1),  # LPR可能随时更新
        }

        logger.info("✅ SimplifiedSyncEngine 初始化成功")

    def sync_indicator(
        self, indicator: str, force_sync: bool = False
    ) -> Dict[str, Any]:
        """
        简化的指标同步逻辑

        Args:
            indicator: 指标名称
            force_sync: 是否强制同步（忽略最小更新间隔）

        Returns:
            Dict: 同步结果
        """
        try:
            logger.info(f"🔄 开始同步 {indicator} 数据...")

            # 检查是否需要同步
            if not force_sync and not self._should_sync(indicator):
                return {
                    "indicator": indicator,
                    "status": "skipped",
                    "reason": "未到最小更新间隔",
                    "synced_count": 0,
                }

            # 获取数据库中最新的时间点
            latest_period = self._get_latest_period_from_db(indicator)

            if latest_period is None:
                # 情况1：数据库为空，全量同步
                start_period = self.start_periods.get(indicator)
                end_period = self._get_current_period(indicator)
                logger.info(f"🔄 {indicator} 全量同步: {start_period} → {end_period}")

            else:
                # 情况2：有数据，增量同步
                start_period = self._get_next_period(latest_period, indicator)
                end_period = self._get_current_period(indicator)

                if self._is_period_greater(start_period, end_period, indicator):
                    logger.info(f"✅ {indicator} 已是最新，无需同步")
                    return {
                        "indicator": indicator,
                        "status": "up_to_date",
                        "reason": "数据已是最新",
                        "synced_count": 0,
                    }

                logger.info(f"🔄 {indicator} 增量同步: {start_period} → {end_period}")

            # 获取并保存数据
            data = self._fetch_data(indicator, start_period, end_period)

            if not data.empty:
                storage = self.storage_manager.get_storage()
                storage.save_data(indicator, data)

                # 更新同步时间戳
                self._update_sync_timestamp(indicator)

                logger.info(f"✅ {indicator} 同步成功: {len(data)} 条记录")
                return {
                    "indicator": indicator,
                    "status": "completed",
                    "synced_count": len(data),
                    "start_period": start_period,
                    "end_period": end_period,
                }
            else:
                logger.warning(f"⚠️ {indicator} 未获取到新数据")
                return {
                    "indicator": indicator,
                    "status": "no_data",
                    "reason": "数据源无新数据",
                    "synced_count": 0,
                }

        except Exception as e:
            logger.error(f"❌ {indicator} 同步失败: {e}")
            return {
                "indicator": indicator,
                "status": "failed",
                "error": str(e),
                "synced_count": 0,
            }

    def _get_latest_period_from_db(self, indicator: str) -> Optional[str]:
        """获取数据库中最新的时间点"""
        try:
            storage = self.storage_manager.get_storage()
            latest_data = storage.get_latest_data(indicator, periods=1)

            if latest_data.empty:
                return None

            # 获取时间字段名
            time_field = INDICATOR_TIME_FIELD.get(indicator)
            if not time_field or time_field not in latest_data.columns:
                logger.error(f"❌ {indicator} 缺少时间字段: {time_field}")
                return None

            # 获取最新的时间点
            latest_period = latest_data[time_field].iloc[0]
            logger.info(f"📊 {indicator} 数据库最新时间点: {latest_period}")
            return str(latest_period)

        except Exception as e:
            logger.error(f"❌ 获取 {indicator} 最新时间点失败: {e}")
            return None

    def _get_current_period(self, indicator: str) -> str:
        """获取考虑发布滞后的当前期间"""
        now = datetime.now()
        delay_config = self.publish_delays.get(indicator, {"type": "month", "delay": 1})

        if delay_config["type"] == "quarter":
            # 季度数据处理
            current_quarter = (now.month - 1) // 3 + 1
            target_quarter = current_quarter - delay_config["delay"]

            if target_quarter <= 0:
                target_quarter += 4
                target_year = now.year - 1
            else:
                target_year = now.year

            return f"{target_year}Q{target_quarter}"

        elif delay_config["type"] == "month":
            # 月度数据处理
            target_month = now.month - delay_config["delay"]

            if target_month <= 0:
                target_month += 12
                target_year = now.year - 1
            else:
                target_year = now.year

            return f"{target_year:04d}{target_month:02d}"

        elif delay_config["type"] == "day":
            # 日度数据处理（主要是LPR）
            target_date = now - timedelta(days=delay_config["delay"])
            return target_date.strftime("%Y%m%d")

        return now.strftime("%Y%m")

    def _get_next_period(self, latest_period: str, indicator: str) -> str:
        """计算下一个期间"""
        frequency = INDICATOR_FREQUENCY.get(indicator, "monthly")

        if frequency == "quarterly":
            # GDP: 2024Q2 → 2024Q3
            year, quarter = self._parse_quarter(latest_period)
            if quarter == 4:
                return f"{year + 1}Q1"
            else:
                return f"{year}Q{quarter + 1}"

        elif frequency == "monthly":
            # CPI: 202409 → 202410
            year, month = self._parse_month(latest_period)
            if month == 12:
                return f"{year + 1:04d}01"
            else:
                return f"{year:04d}{month + 1:02d}"

        elif frequency == "irregular":
            # LPR: 特殊处理，从最新日期开始
            return latest_period

        return latest_period

    def _parse_quarter(self, period: str) -> tuple:
        """解析季度字符串: 2024Q2 → (2024, 2)"""
        try:
            year_str, quarter_str = period.split("Q")
            return int(year_str), int(quarter_str)
        except ValueError:
            logger.error(f"❌ 无法解析季度格式: {period}")
            return datetime.now().year, 1

    def _parse_month(self, period: str) -> tuple:
        """解析月份字符串: 202409 → (2024, 9)"""
        try:
            period_str = str(period)
            if len(period_str) == 6:
                year = int(period_str[:4])
                month = int(period_str[4:6])
                return year, month
            else:
                raise ValueError(f"月份格式错误: {period}")
        except ValueError:
            logger.error(f"❌ 无法解析月份格式: {period}")
            return datetime.now().year, datetime.now().month

    def _is_period_greater(self, period1: str, period2: str, indicator: str) -> bool:
        """比较两个时间段的大小"""
        frequency = INDICATOR_FREQUENCY.get(indicator, "monthly")

        if frequency == "quarterly":
            year1, quarter1 = self._parse_quarter(period1)
            year2, quarter2 = self._parse_quarter(period2)
            return (year1, quarter1) > (year2, quarter2)

        elif frequency == "monthly":
            year1, month1 = self._parse_month(period1)
            year2, month2 = self._parse_month(period2)
            return (year1, month1) > (year2, month2)

        elif frequency == "irregular":
            # 日期比较
            try:
                date1 = datetime.strptime(period1, "%Y%m%d")
                date2 = datetime.strptime(period2, "%Y%m%d")
                return date1 > date2
            except ValueError:
                return period1 > period2

        return period1 > period2

    def _fetch_data(
        self, indicator: str, start_period: str, end_period: str
    ) -> pd.DataFrame:
        """获取指定时间范围的数据"""
        try:
            logger.info(f"🔄 获取 {indicator} 数据: {start_period} → {end_period}")

            # 特殊处理LPR的不定期发布
            if indicator == "lpr":
                return self._fetch_lpr_special(start_period, end_period)

            # 根据指标类型调用对应的获取方法
            if indicator == "gdp":
                return self.fetcher.fetch_gdp(start_q=start_period, end_q=end_period)
            elif indicator == "cpi":
                return self.fetcher.fetch_cpi(start_m=start_period, end_m=end_period)
            elif indicator == "ppi":
                return self.fetcher.fetch_ppi(start_m=start_period, end_m=end_period)
            elif indicator == "pmi":
                return self.fetcher.fetch_pmi(start_m=start_period, end_m=end_period)
            elif indicator == "money_supply":
                return self.fetcher.fetch_money_supply(
                    start_m=start_period, end_m=end_period
                )
            elif indicator == "social_financing":
                return self.fetcher.fetch_social_financing(
                    start_m=start_period, end_m=end_period
                )
            else:
                logger.error(f"❌ 未知指标: {indicator}")
                return pd.DataFrame()

        except Exception as e:
            logger.error(f"❌ 获取 {indicator} 数据失败: {e}")
            return pd.DataFrame()

    def _fetch_lpr_special(self, start_period: str, end_period: str) -> pd.DataFrame:
        """LPR的特殊处理：不定期发布"""
        try:
            # LPR不定期发布，获取最近3个月的数据让Tushare自动过滤
            three_months_ago = (datetime.now() - timedelta(days=90)).strftime("%Y%m%d")
            current_date = datetime.now().strftime("%Y%m%d")

            logger.info(
                f"🔄 LPR特殊处理: 获取最近3个月数据 "
                f"{three_months_ago} → {current_date}"
            )

            all_data = self.fetcher.fetch_lpr(
                start_date=three_months_ago, end_date=current_date
            )

            if all_data.empty:
                return all_data

            # 过滤掉已经存在的数据（比start_period更新的数据）
            if "date" in all_data.columns:
                # 只保留比start_period更新的数据
                all_data = all_data[all_data["date"] >= start_period]

            return all_data

        except Exception as e:
            logger.error(f"❌ LPR特殊处理失败: {e}")
            return pd.DataFrame()

    def _should_sync(self, indicator: str) -> bool:
        """检查是否应该同步指定指标"""
        # 检查最小更新间隔
        last_sync = self._get_last_sync_timestamp(indicator)
        if last_sync:
            min_interval = self.min_sync_interval.get(indicator, timedelta(days=1))
            if datetime.now() - last_sync < min_interval:
                return False
        return True

    def detect_missing_periods(self, indicator: str) -> List[str]:
        """
        检测指定指标的缺失时间段（保留此方法用于兼容性和健康检查）
        注意：这个方法现在主要用于健康检查，核心同步逻辑已简化
        """
        try:
            latest_period = self._get_latest_period_from_db(indicator)
            current_period = self._get_current_period(indicator)

            if latest_period is None:
                start_period = self.start_periods.get(indicator)
                return [f"需要全量同步从 {start_period} 到 {current_period}"]

            if self._is_period_greater(latest_period, current_period, indicator):
                return []  # 无缺失

            next_period = self._get_next_period(latest_period, indicator)
            return [f"需要增量同步从 {next_period} 到 {current_period}"]

        except Exception as e:
            logger.error(f"❌ 检测 {indicator} 缺失时间段失败: {e}")
            return []

    def _get_last_sync_timestamp(self, indicator: str) -> Optional[datetime]:
        """获取指标的最后同步时间戳"""
        # TODO: 实现同步时间戳的持久化存储
        # 暂时返回None，表示总是需要检查
        return None

    def _update_sync_timestamp(self, indicator: str):
        """更新指标的同步时间戳"""
        # TODO: 实现同步时间戳的持久化存储
        pass

    def sync_all_indicators(self, force_sync: bool = False) -> Dict[str, Any]:
        """
        同步所有指标数据

        Args:
            force_sync: 是否强制同步

        Returns:
            Dict: 同步结果汇总
        """
        indicators = [
            "gdp",
            "cpi",
            "ppi",
            "pmi",
            "money_supply",
            "social_financing",
            "lpr",
        ]

        results = {}
        total_synced = 0

        logger.info(f"🚀 开始同步所有宏观数据指标 ({len(indicators)} 个)...")

        for indicator in indicators:
            try:
                result = self.sync_indicator(indicator, force_sync=force_sync)
                results[indicator] = result
                total_synced += result.get("synced_count", 0)

                # 指标间稍作延迟，避免API限制
                time.sleep(1)

            except Exception as e:
                logger.error(f"❌ 同步 {indicator} 时发生错误: {e}")
                results[indicator] = {
                    "status": "failed",
                    "error": str(e),
                    "synced_count": 0,
                }

        # 汇总结果
        summary = {
            "total_indicators": len(indicators),
            "completed": len(
                [r for r in results.values() if r.get("status") == "completed"]
            ),
            "partial": len(
                [r for r in results.values() if r.get("status") == "partial"]
            ),
            "failed": len([r for r in results.values() if r.get("status") == "failed"]),
            "skipped": len(
                [r for r in results.values() if r.get("status") == "skipped"]
            ),
            "up_to_date": len(
                [r for r in results.values() if r.get("status") == "up_to_date"]
            ),
            "total_synced_records": total_synced,
            "results": results,
        }

        logger.info(f"🎯 全量同步完成: {summary}")
        return summary