import time

import akshare as ak
import pandas as pd

from config import DATA_CACHE_HOURS, CACHE_DIR
from data.cache import get_cached, set_cached

SNAPSHOT_PREFIX = "snapshot_"


def save_snapshot(df: pd.DataFrame):
    """保存当日真实快照（仅成交额sum>0时保存），供次日盘前候选池使用"""
    if df is None or df.empty:
        return
    amt_col = "成交额(元)" if "成交额(元)" in df.columns else None
    if amt_col is not None and df[amt_col].sum() <= 0:
        return
    date = pd.Timestamp.now().strftime("%Y%m%d")
    path = CACHE_DIR / f"{SNAPSHOT_PREFIX}{date}.csv"
    df.to_csv(path, index=False, encoding="utf-8-sig")


def load_last_snapshot(max_age_days: int = 10) -> pd.DataFrame:
    """加载最近一份有真实成交额的历史快照，用于盘前候选池排序"""
    files = sorted(CACHE_DIR.glob(f"{SNAPSHOT_PREFIX}*.csv"), reverse=True)
    for f in files:
        age_days = (time.time() - f.stat().st_mtime) / 86400
        if age_days > max_age_days:
            continue
        try:
            df = pd.read_csv(f)
            if "代码" in df.columns and "成交额(元)" in df.columns \
               and df["成交额(元)"].sum() > 0:
                df["代码"] = df["代码"].map(_normalize_code)
                return df
        except Exception:
            continue
    return pd.DataFrame()


def _normalize_code(code) -> str:
    """将快照CSV中的股票代码规范化为6位字符串"""
    s = str(code).strip()
    if s.endswith(".0"):
        s = s[:-2]
    return s.zfill(6) if s.isdigit() else s


def get_market_snapshot() -> pd.DataFrame:
    key = "market_snapshot"
    cached = get_cached(key, DATA_CACHE_HOURS)
    if cached is not None:
        return cached
    try:
        df = ak.stock_zh_a_spot_em()
        set_cached(key, df)
        return df
    except Exception:
        pass
    from data.sina_fetcher import sina_get_market_snapshot
    df = sina_get_market_snapshot()
    if not df.empty:
        set_cached(key, df)
    return df


def get_stock_hist(symbol: str, days: int = 60, period: str = "daily") -> pd.DataFrame:
    key = f"hist_{symbol}_{period}_{days}"
    cached = get_cached(key, DATA_CACHE_HOURS)
    if cached is not None:
        return cached
    end = pd.Timestamp.now().strftime("%Y%m%d")
    start = (pd.Timestamp.now() - pd.Timedelta(days=days * 2)).strftime("%Y%m%d")
    try:
        df = ak.stock_zh_a_hist(
            symbol=symbol, period=period,
            start_date=start, end_date=end, adjust="qfq"
        )
        if df is not None and not df.empty:
            set_cached(key, df)
            return df
    except Exception:
        pass
    from data.sina_fetcher import sina_get_stock_hist
    df = sina_get_stock_hist(symbol, days)
    if not df.empty:
        set_cached(key, df)
    return df


def get_stock_min(symbol: str, period: str = "5", days: int = 5) -> pd.DataFrame:
    key = f"min_{symbol}_{period}"
    cached = get_cached(key, 1)
    if cached is not None:
        return cached
    end = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
    start = (pd.Timestamp.now() - pd.Timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
    try:
        df = ak.stock_zh_a_hist_min_em(
            symbol=symbol, period=period,
            start_date=start, end_date=end, adjust=""
        )
        if df is not None and not df.empty:
            set_cached(key, df)
        return df
    except Exception:
        return pd.DataFrame()


def get_limit_up_pool(date: str = None) -> pd.DataFrame:
    if date is None:
        date = pd.Timestamp.now().strftime("%Y%m%d")
    key = f"zt_pool_{date}"
    cached = get_cached(key, 8)
    if cached is not None:
        return cached
    try:
        df = ak.stock_zt_pool_em(date=date)
        if df is not None and not df.empty:
            set_cached(key, df)
        return df
    except Exception:
        return pd.DataFrame()


def get_fund_flow_rank(indicator: str = "今日") -> pd.DataFrame:
    key = f"fund_flow_rank_{indicator}"
    cached = get_cached(key, DATA_CACHE_HOURS)
    if cached is not None:
        return cached
    try:
        df = ak.stock_individual_fund_flow_rank(indicator=indicator)
        if df is not None and not df.empty:
            set_cached(key, df)
        return df
    except Exception:
        return pd.DataFrame()


def get_industry_boards() -> pd.DataFrame:
    key = "industry_boards"
    cached = get_cached(key, DATA_CACHE_HOURS)
    if cached is not None:
        return cached
    try:
        df = ak.stock_board_industry_name_em()
        if df is not None and not df.empty:
            set_cached(key, df)
        return df
    except Exception:
        return pd.DataFrame()


def get_concept_boards() -> pd.DataFrame:
    key = "concept_boards"
    cached = get_cached(key, DATA_CACHE_HOURS)
    if cached is not None:
        return cached
    try:
        df = ak.stock_board_concept_name_em()
        if df is not None and not df.empty:
            set_cached(key, df)
        return df
    except Exception:
        return pd.DataFrame()


def get_stock_fund_flow(symbol: str, market: str = "sh") -> pd.DataFrame:
    key = f"fund_flow_{symbol}"
    cached = get_cached(key, DATA_CACHE_HOURS)
    if cached is not None:
        return cached
    try:
        df = ak.stock_individual_fund_flow(stock=symbol, market=market)
        if df is not None and not df.empty:
            set_cached(key, df)
        return df
    except Exception:
        return pd.DataFrame()


def batch_stock_hist(symbols: list, days: int = 60, period: str = "daily",
                     sleep: float = 0.5) -> dict:
    result = {}
    for i, sym in enumerate(symbols):
        df = get_stock_hist(sym, days, period)
        if df is not None and not df.empty:
            result[sym] = df
        if i < len(symbols) - 1:
            time.sleep(sleep)
    return result