Akshare

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# Virtual Environment
venv/
ENV/
env/

# IDE
.vscode/
.idea/
*.swp
*.swo
*~

# OS
.DS_Store
Thumbs.db

# Testing
.pytest_cache/
.coverage
htmlcov/

# Logs
*.log

# Data files (optional - uncomment if you want to ignore data files)
# data/
# *.csv
# *.xlsx

MIT License

Copyright (c) 2025 OpenClaw Contributors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AkShare Financial Data Tool
获取中国金融市场数据的命令行工具

功能：
- A股实时行情（支持筛选）
- 股票历史数据（日线/周线/月线）
- 港股/美股数据
- 期货数据
- 基金数据
- 宏观经济数据（GDP/CPI/PMI）
- 指数数据
"""

import argparse
import sys
import json
import time
import os
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Dict, Any, List
import hashlib

try:
    import akshare as ak
    import pandas as pd
except ImportError as e:
    print(f"错误: 缺少必要的依赖库")
    print(f"请运行: pip install akshare pandas")
    sys.exit(1)


class CacheManager:
    """数据缓存管理器"""
    
    def __init__(self, cache_dir: str = None, cache_expiry_hours: int = 24):
        """
        初始化缓存管理器
        
        Args:
            cache_dir: 缓存目录路径
            cache_expiry_hours: 缓存过期时间（小时）
        """
        if cache_dir is None:
            cache_dir = os.path.join(os.path.expanduser("~"), ".akshare_cache")
        
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.cache_expiry_hours = cache_expiry_hours
    
    def _get_cache_key(self, func_name: str, **kwargs) -> str:
        """生成缓存键"""
        # 将参数转换为字符串并排序
        params_str = json.dumps(kwargs, sort_keys=True, ensure_ascii=False)
        # 使用MD5生成唯一键
        key_str = f"{func_name}:{params_str}"
        return hashlib.md5(key_str.encode('utf-8')).hexdigest()
    
    def get(self, func_name: str, **kwargs) -> Optional[pd.DataFrame]:
        """
        从缓存获取数据
        
        Args:
            func_name: 函数名
            **kwargs: 函数参数
            
        Returns:
            缓存的DataFrame，如果不存在或已过期则返回None
        """
        cache_key = self._get_cache_key(func_name, **kwargs)
        cache_file = self.cache_dir / f"{cache_key}.parquet"
        
        if not cache_file.exists():
            return None
        
        # 检查缓存是否过期
        file_time = datetime.fromtimestamp(cache_file.stat().st_mtime)
        expiry_time = datetime.now() - timedelta(hours=self.cache_expiry_hours)
        
        if file_time < expiry_time:
            cache_file.unlink()  # 删除过期缓存
            return None
        
        try:
            df = pd.read_parquet(cache_file)
            return df
        except Exception:
            return None
    
    def set(self, func_name: str, data: pd.DataFrame, **kwargs) -> bool:
        """
        保存数据到缓存
        
        Args:
            func_name: 函数名
            data: 要缓存的数据
            **kwargs: 函数参数
            
        Returns:
            是否成功保存
        """
        cache_key = self._get_cache_key(func_name, **kwargs)
        cache_file = self.cache_dir / f"{cache_key}.parquet"
        
        try:
            data.to_parquet(cache_file)
            return True
        except Exception as e:
            print(f"缓存保存失败: {e}")
            return False
    
    def clear(self) -> int:
        """
        清除所有缓存
        
        Returns:
            删除的缓存文件数量
        """
        count = 0
        for file in self.cache_dir.glob("*.parquet"):
            file.unlink()
            count += 1
        return count


def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """
    失败重试装饰器
    
    Args:
        max_retries: 最大重试次数
        delay: 重试延迟（秒）
    """
    def decorator(func):
        def wrapper(*args, **kwargs):
            last_error = None
            current_delay = delay  # 使用局部变量
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_retries - 1:
                        print(f"请求失败，{current_delay}秒后重试... (尝试 {attempt + 1}/{max_retries})")
                        time.sleep(current_delay)
                        current_delay *= 2  # 指数退避
            raise last_error
        return wrapper
    return decorator


class AkShareTool:
    """AkShare 数据获取工具"""
    
    def __init__(self, use_cache: bool = True, cache_expiry_hours: int = 24):
        """
        初始化工具
        
        Args:
            use_cache: 是否使用缓存
            cache_expiry_hours: 缓存过期时间（小时）
        """
        self.use_cache = use_cache
        self.cache = CacheManager(cache_expiry_hours=cache_expiry_hours) if use_cache else None
    
    def _fetch_with_cache(self, func_name: str, fetch_func, use_cache: bool = None, **kwargs) -> pd.DataFrame:
        """
        带缓存的数据获取
        
        Args:
            func_name: 函数名（用于缓存键）
            fetch_func: 实际获取数据的函数（无参数）
            use_cache: 是否使用缓存（覆盖默认设置）
            **kwargs: 函数参数（仅用于缓存键）
            
        Returns:
            DataFrame数据
        """
        should_cache = use_cache if use_cache is not None else self.use_cache
        
        # 尝试从缓存获取
        if should_cache and self.cache:
            cached_data = self.cache.get(func_name, **kwargs)
            if cached_data is not None:
                print("✓ 使用缓存数据")
                return cached_data
        
        # 从网络获取
        print("正在获取数据...")
        data = fetch_func()
        
        # 保存到缓存
        if should_cache and self.cache and data is not None and not data.empty:
            self.cache.set(func_name, data, **kwargs)
            print("✓ 数据已缓存")
        
        return data
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_stock_realtime(self, symbol: str = None, filter_field: str = None, 
                          filter_value: str = None, use_cache: bool = False) -> pd.DataFrame:
        """
        获取A股实时行情
        
        Args:
            symbol: 股票代码（可选，为空则获取全部）
            filter_field: 筛选字段（如 '行业', '市场' 等）
            filter_value: 筛选值
            use_cache: 是否使用缓存
            
        Returns:
            实时行情数据
        """
        def fetch():
            if symbol:
                # 单个股票
                df = ak.stock_zh_a_spot()
                return df[df['代码'] == symbol] if '代码' in df.columns else df
            else:
                # 全部A股
                df = ak.stock_zh_a_spot_em()
                # 筛选
                if filter_field and filter_value and filter_field in df.columns:
                    df = df[df[filter_field].astype(str).str.contains(filter_value, na=False)]
                return df
        
        return self._fetch_with_cache("stock_realtime", fetch, use_cache=use_cache, 
                                     symbol=symbol, filter_field=filter_field, filter_value=filter_value)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_stock_history(self, symbol: str, period: str = "daily", 
                         start_date: str = None, end_date: str = None,
                         adjust: str = "qfq", use_cache: bool = True) -> pd.DataFrame:
        """
        获取股票历史数据
        
        Args:
            symbol: 股票代码（如 "000001"）
            period: 周期（daily/weekly/monthly）
            start_date: 开始日期（格式：YYYYMMDD）
            end_date: 结束日期（格式：YYYYMMDD）
            adjust: 复权类型（qfq-前复权, hfq-后复权, ""-不复权）
            use_cache: 是否使用缓存
            
        Returns:
            历史数据
        """
        # 默认日期范围
        if not end_date:
            end_date = datetime.now().strftime("%Y%m%d")
        if not start_date:
            start_date = (datetime.now() - timedelta(days=365)).strftime("%Y%m%d")
        
        def fetch():
            df = ak.stock_zh_a_hist(
                symbol=symbol,
                period=period,
                start_date=start_date,
                end_date=end_date,
                adjust=adjust
            )
            return df
        
        return self._fetch_with_cache("stock_history", fetch, use_cache=use_cache,
                                     symbol=symbol, period=period, start_date=start_date,
                                     end_date=end_date, adjust=adjust)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_hk_stock_realtime(self, use_cache: bool = False) -> pd.DataFrame:
        """获取港股实时行情"""
        def fetch():
            return ak.stock_hk_spot_em()
        
        return self._fetch_with_cache("hk_stock_realtime", fetch, use_cache=use_cache)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_hk_stock_history(self, symbol: str, period: str = "daily",
                            start_date: str = None, end_date: str = None,
                            adjust: str = "qfq", use_cache: bool = True) -> pd.DataFrame:
        """获取港股历史数据"""
        if not end_date:
            end_date = datetime.now().strftime("%Y%m%d")
        if not start_date:
            start_date = (datetime.now() - timedelta(days=365)).strftime("%Y%m%d")
        
        def fetch():
            df = ak.stock_hk_hist(
                symbol=symbol,
                period=period,
                start_date=start_date,
                end_date=end_date,
                adjust=adjust
            )
            return df
        
        return self._fetch_with_cache("hk_stock_history", fetch, use_cache=use_cache,
                                     symbol=symbol, period=period, start_date=start_date,
                                     end_date=end_date, adjust=adjust)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_us_stock_realtime(self, use_cache: bool = False) -> pd.DataFrame:
        """获取美股实时行情"""
        def fetch():
            return ak.stock_us_spot_em()
        
        return self._fetch_with_cache("us_stock_realtime", fetch, use_cache=use_cache)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_futures_realtime(self, symbol: str = None, use_cache: bool = False) -> pd.DataFrame:
        """
        获取期货实时行情
        
        Args:
            symbol: 期货代码（可选）
            use_cache: 是否使用缓存
        """
        def fetch():
            df = ak.futures_zh_spot()
            if symbol and '代码' in df.columns:
                df = df[df['代码'].str.contains(symbol, na=False)]
            return df
        
        return self._fetch_with_cache("futures_realtime", fetch, use_cache=use_cache, symbol=symbol)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_futures_history(self, symbol: str, start_date: str = None,
                           end_date: str = None, use_cache: bool = True) -> pd.DataFrame:
        """获取期货历史数据"""
        if not end_date:
            end_date = datetime.now().strftime("%Y%m%d")
        if not start_date:
            start_date = (datetime.now() - timedelta(days=90)).strftime("%Y%m%d")
        
        def fetch():
            df = ak.futures_zh_hist_sina(symbol=symbol)
            # 日期筛选
            if 'date' in df.columns:
                df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]
            return df
        
        return self._fetch_with_cache("futures_history", fetch, use_cache=use_cache,
                                     symbol=symbol, start_date=start_date, end_date=end_date)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_fund_list(self, fund_type: str = None, use_cache: bool = True) -> pd.DataFrame:
        """
        获取基金列表
        
        Args:
            fund_type: 基金类型（可选）
            use_cache: 是否使用缓存
        """
        def fetch():
            df = ak.fund_open_fund_info_em()
            if fund_type and '基金类型' in df.columns:
                df = df[df['基金类型'].str.contains(fund_type, na=False)]
            return df
        
        return self._fetch_with_cache("fund_list", fetch, use_cache=use_cache, fund_type=fund_type)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_fund_history(self, fund_code: str, indicator: str = "单位净值走势",
                        use_cache: bool = True) -> pd.DataFrame:
        """
        获取基金历史数据
        
        Args:
            fund_code: 基金代码
            indicator: 指标类型（单位净值走势/累计净值走势等）
            use_cache: 是否使用缓存
        """
        def fetch():
            df = ak.fund_open_fund_info_em(fund=fund_code, indicator=indicator)
            return df
        
        return self._fetch_with_cache("fund_history", fetch, use_cache=use_cache,
                                     fund_code=fund_code, indicator=indicator)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_macro_gdp(self, use_cache: bool = True) -> pd.DataFrame:
        """获取GDP数据"""
        def fetch():
            return ak.macro_china_gdp()
        
        return self._fetch_with_cache("macro_gdp", fetch, use_cache=use_cache)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_macro_cpi(self, use_cache: bool = True) -> pd.DataFrame:
        """获取CPI数据"""
        def fetch():
            return ak.macro_china_cpi()
        
        return self._fetch_with_cache("macro_cpi", fetch, use_cache=use_cache)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_macro_pmi(self, use_cache: bool = True) -> pd.DataFrame:
        """获取PMI数据"""
        def fetch():
            return ak.macro_china_pmi()
        
        return self._fetch_with_cache("macro_pmi", fetch, use_cache=use_cache)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_index_realtime(self, index_type: str = "all", use_cache: bool = False) -> pd.DataFrame:
        """
        获取指数实时行情
        
        Args:
            index_type: 指数类型（all/sz/sh）
            use_cache: 是否使用缓存
        """
        def fetch():
            if index_type == "sz":
                df = ak.index_stock_info_sz_name_code()
            elif index_type == "sh":
                df = ak.index_stock_info_sh_name_code()
            else:
                # 获取主要指数实时行情
                df = ak.index_zh_a_spot()
            return df
        
        return self._fetch_with_cache("index_realtime", fetch, use_cache=use_cache, index_type=index_type)
    
    @retry_on_failure(max_retries=3, delay=1.0)
    def get_index_history(self, symbol: str, period: str = "daily",
                         start_date: str = None, end_date: str = None,
                         use_cache: bool = True) -> pd.DataFrame:
        """获取指数历史数据"""
        if not end_date:
            end_date = datetime.now().strftime("%Y%m%d")
        if not start_date:
            start_date = (datetime.now() - timedelta(days=365)).strftime("%Y%m%d")
        
        def fetch():
            df = ak.index_zh_a_hist(
                symbol=symbol,
                period=period,
                start_date=start_date,
                end_date=end_date
            )
            return df
        
        return self._fetch_with_cache("index_history", fetch, use_cache=use_cache,
                                     symbol=symbol, period=period, start_date=start_date,
                                     end_date=end_date)
    
    def clear_cache(self) -> int:
        """清除缓存"""
        if self.cache:
            return self.cache.clear()
        return 0


def print_dataframe(df: pd.DataFrame, max_rows: int = 20, title: str = "数据结果"):
    """美化打印DataFrame"""
    if df is None or df.empty:
        print("没有数据")
        return
    
    print(f"\n{'='*80}")
    print(f"  {title}")
    print(f"  共 {len(df)} 条记录")
    print(f"{'='*80}\n")
    
    # 设置显示选项
    pd.set_option('display.max_columns', None)
    pd.set_option('display.width', None)
    pd.set_option('display.max_colwidth', 20)
    
    # 分页显示
    if len(df) > max_rows:
        print(df.head(max_rows))
        print(f"\n... (还有 {len(df) - max_rows} 条记录)")
    else:
        print(df)
    
    print("\n" + "="*80)


def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description="AkShare 金融市场数据获取工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 获取所有A股实时行情
  %(prog)s stock-realtime
  
  # 获取单只股票实时行情
  %(prog)s stock-realtime --symbol 000001
  
  # 筛选特定行业的股票
  %(prog)s stock-realtime --filter-field 行业 --filter-value 银行
  
  # 获取股票历史数据
  %(prog)s stock-history --symbol 000001 --period daily --adjust qfq
  
  # 获取港股实时行情
  %(prog)s hk-realtime
  
  # 获取基金列表
  %(prog)s fund-list
  
  # 获取宏观GDP数据
  %(prog)s macro-gdp
  
  # 清除缓存
  %(prog)s clear-cache
        """
    )
    
    # 全局参数
    parser.add_argument("--no-cache", action="store_true", help="不使用缓存")
    parser.add_argument("--cache-expiry", type=int, default=24, 
                       help="缓存过期时间（小时）")
    parser.add_argument("--output", "-o", type=str, help="输出文件路径（CSV/JSON格式）")
    parser.add_argument("--format", type=str, choices=["table", "json", "csv"], 
                       default="table", help="输出格式")
    
    # 子命令
    subparsers = parser.add_subparsers(dest="command", help="数据类型")
    
    # A股实时行情
    stock_realtime_parser = subparsers.add_parser("stock-realtime", help="A股实时行情")
    stock_realtime_parser.add_argument("--symbol", "-s", type=str, help="股票代码")
    stock_realtime_parser.add_argument("--filter-field", type=str, help="筛选字段")
    stock_realtime_parser.add_argument("--filter-value", type=str, help="筛选值")
    
    # 股票历史数据
    stock_history_parser = subparsers.add_parser("stock-history", help="股票历史数据")
    stock_history_parser.add_argument("--symbol", "-s", type=str, required=True, help="股票代码")
    stock_history_parser.add_argument("--period", "-p", type=str, default="daily",
                                     choices=["daily", "weekly", "monthly"], help="周期")
    stock_history_parser.add_argument("--start-date", type=str, help="开始日期（YYYYMMDD）")
    stock_history_parser.add_argument("--end-date", type=str, help="结束日期（YYYYMMDD）")
    stock_history_parser.add_argument("--adjust", type=str, default="qfq",
                                     choices=["qfq", "hfq", ""], help="复权类型")
    
    # 港股实时行情
    subparsers.add_parser("hk-realtime", help="港股实时行情")
    
    # 港股历史数据
    hk_history_parser = subparsers.add_parser("hk-history", help="港股历史数据")
    hk_history_parser.add_argument("--symbol", "-s", type=str, required=True, help="股票代码")
    hk_history_parser.add_argument("--period", "-p", type=str, default="daily",
                                   choices=["daily", "weekly", "monthly"], help="周期")
    hk_history_parser.add_argument("--start-date", type=str, help="开始日期（YYYYMMDD）")
    hk_history_parser.add_argument("--end-date", type=str, help="结束日期（YYYYMMDD）")
    hk_history_parser.add_argument("--adjust", type=str, default="qfq",
                                   choices=["qfq", "hfq", ""], help="复权类型")
    
    # 美股实时行情
    subparsers.add_parser("us-realtime", help="美股实时行情")
    
    # 期货实时行情
    futures_realtime_parser = subparsers.add_parser("futures-realtime", help="期货实时行情")
    futures_realtime_parser.add_argument("--symbol", "-s", type=str, help="期货代码")
    
    # 期货历史数据
    futures_history_parser = subparsers.add_parser("futures-history", help="期货历史数据")
    futures_history_parser.add_argument("--symbol", "-s", type=str, required=True, help="期货代码")
    futures_history_parser.add_argument("--start-date", type=str, help="开始日期（YYYYMMDD）")
    futures_history_parser.add_argument("--end-date", type=str, help="结束日期（YYYYMMDD）")
    
    # 基金列表
    fund_list_parser = subparsers.add_parser("fund-list", help="基金列表")
    fund_list_parser.add_argument("--type", type=str, help="基金类型")
    
    # 基金历史数据
    fund_history_parser = subparsers.add_parser("fund-history", help="基金历史数据")
    fund_history_parser.add_argument("--code", "-c", type=str, required=True, help="基金代码")
    fund_history_parser.add_argument("--indicator", type=str, default="单位净值走势", help="指标类型")
    
    # 宏观经济数据
    subparsers.add_parser("macro-gdp", help="GDP数据")
    subparsers.add_parser("macro-cpi", help="CPI数据")
    subparsers.add_parser("macro-pmi", help="PMI数据")
    
    # 指数实时行情
    index_realtime_parser = subparsers.add_parser("index-realtime", help="指数实时行情")
    index_realtime_parser.add_argument("--type", type=str, default="all",
                                      choices=["all", "sz", "sh"], help="指数类型")
    
    # 指数历史数据
    index_history_parser = subparsers.add_parser("index-history", help="指数历史数据")
    index_history_parser.add_argument("--symbol", "-s", type=str, required=True, help="指数代码")
    index_history_parser.add_argument("--period", "-p", type=str, default="daily",
                                     choices=["daily", "weekly", "monthly"], help="周期")
    index_history_parser.add_argument("--start-date", type=str, help="开始日期（YYYYMMDD）")
    index_history_parser.add_argument("--end-date", type=str, help="结束日期（YYYYMMDD）")
    
    # 清除缓存
    subparsers.add_parser("clear-cache", help="清除缓存")
    
    args = parser.parse_args()
    
    # 如果没有命令，显示帮助
    if not args.command:
        parser.print_help()
        return
    
    # 初始化工具
    tool = AkShareTool(use_cache=not args.no_cache, cache_expiry_hours=args.cache_expiry)
    
    # 执行命令
    df = None
    title = "数据结果"
    
    try:
        if args.command == "stock-realtime":
            df = tool.get_stock_realtime(
                symbol=args.symbol,
                filter_field=args.filter_field,
                filter_value=args.filter_value,
                use_cache=not args.no_cache
            )
            title = f"A股实时行情 - {args.symbol if args.symbol else '全部'}"
        
        elif args.command == "stock-history":
            df = tool.get_stock_history(
                symbol=args.symbol,
                period=args.period,
                start_date=args.start_date,
                end_date=args.end_date,
                adjust=args.adjust,
                use_cache=not args.no_cache
            )
            title = f"股票历史数据 - {args.symbol} ({args.period})"
        
        elif args.command == "hk-realtime":
            df = tool.get_hk_stock_realtime(use_cache=not args.no_cache)
            title = "港股实时行情"
        
        elif args.command == "hk-history":
            df = tool.get_hk_stock_history(
                symbol=args.symbol,
                period=args.period,
                start_date=args.start_date,
                end_date=args.end_date,
                adjust=args.adjust,
                use_cache=not args.no_cache
            )
            title = f"港股历史数据 - {args.symbol} ({args.period})"
        
        elif args.command == "us-realtime":
            df = tool.get_us_stock_realtime(use_cache=not args.no_cache)
            title = "美股实时行情"
        
        elif args.command == "futures-realtime":
            df = tool.get_futures_realtime(
                symbol=args.symbol,
                use_cache=not args.no_cache
            )
            title = f"期货实时行情 - {args.symbol if args.symbol else '全部'}"
        
        elif args.command == "futures-history":
            df = tool.get_futures_history(
                symbol=args.symbol,
                start_date=args.start_date,
                end_date=args.end_date,
                use_cache=not args.no_cache
            )
            title = f"期货历史数据 - {args.symbol}"
        
        elif args.command == "fund-list":
            df = tool.get_fund_list(
                fund_type=args.type,
                use_cache=not args.no_cache
            )
            title = f"基金列表 - {args.type if args.type else '全部'}"
        
        elif args.command == "fund-history":
            df = tool.get_fund_history(
                fund_code=args.code,
                indicator=args.indicator,
                use_cache=not args.no_cache
            )
            title = f"基金历史数据 - {args.code}"
        
        elif args.command == "macro-gdp":
            df = tool.get_macro_gdp(use_cache=not args.no_cache)
            title = "GDP数据"
        
        elif args.command == "macro-cpi":
            df = tool.get_macro_cpi(use_cache=not args.no_cache)
            title = "CPI数据"
        
        elif args.command == "macro-pmi":
            df = tool.get_macro_pmi(use_cache=not args.no_cache)
            title = "PMI数据"
        
        elif args.command == "index-realtime":
            df = tool.get_index_realtime(
                index_type=args.type,
                use_cache=not args.no_cache
            )
            title = f"指数实时行情 - {args.type}"
        
        elif args.command == "index-history":
            df = tool.get_index_history(
                symbol=args.symbol,
                period=args.period,
                start_date=args.start_date,
                end_date=args.end_date,
                use_cache=not args.no_cache
            )
            title = f"指数历史数据 - {args.symbol} ({args.period})"
        
        elif args.command == "clear-cache":
            count = tool.clear_cache()
            print(f"✓ 已清除 {count} 个缓存文件")
            return
        
        # 输出结果
        if df is not None:
            if args.output:
                # 保存到文件
                if args.output.endswith('.json'):
                    df.to_json(args.output, orient='records', force_ascii=False, indent=2)
                else:
                    df.to_csv(args.output, index=False, encoding='utf-8-sig')
                print(f"✓ 数据已保存到: {args.output}")
            elif args.format == "json":
                print(df.to_json(orient='records', force_ascii=False, indent=2))
            elif args.format == "csv":
                print(df.to_csv(index=False))
            else:
                print_dataframe(df, title=title)
    
    except KeyboardInterrupt:
        print("\n\n操作已取消")
        sys.exit(0)
    except Exception as e:
        print(f"\n错误: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AkShare Tool 使用示例
演示如何使用 akshare_tool.py 中的功能
"""

from akshare_tool import AkShareTool
import pandas as pd


def example_stock_realtime():
    """示例1: 获取A股实时行情"""
    print("\n" + "="*80)
    print("示例1: 获取A股实时行情")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取单只股票实时行情
    df = tool.get_stock_realtime(symbol="000001", use_cache=False)
    print(f"\n平安银行(000001)实时行情:")
    print(df)
    
    # 获取银行业股票
    df_banks = tool.get_stock_realtime(filter_field="行业", filter_value="银行", use_cache=False)
    print(f"\n银行板块股票数量: {len(df_banks)}")
    print(df_banks.head(10))


def example_stock_history():
    """示例2: 获取股票历史数据"""
    print("\n" + "="*80)
    print("示例2: 获取股票历史数据")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取平安银行日线数据
    df = tool.get_stock_history(
        symbol="000001",
        period="daily",
        start_date="20240101",
        end_date="20241231",
        adjust="qfq",
        use_cache=True
    )
    
    print(f"\n平安银行历史数据 (最近10天):")
    print(df.tail(10))
    
    # 计算技术指标
    df['MA5'] = df['收盘'].rolling(window=5).mean()
    df['MA20'] = df['收盘'].rolling(window=20).mean()
    
    print(f"\n包含移动平均线的数据 (最近5天):")
    print(df[['日期', '收盘', 'MA5', 'MA20']].tail(5))


def example_hk_stock():
    """示例3: 获取港股数据"""
    print("\n" + "="*80)
    print("示例3: 获取港股数据")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取腾讯控股实时行情
    df = tool.get_hk_stock_realtime(use_cache=False)
    print(f"\n港股实时行情数量: {len(df)}")
    print(df.head(10))
    
    # 获取腾讯控股历史数据
    df_history = tool.get_hk_stock_history(
        symbol="00700",
        period="daily",
        start_date="20240101",
        end_date="20241231",
        use_cache=True
    )
    print(f"\n腾讯控股历史数据 (最近5天):")
    print(df_history.tail(5))


def example_futures():
    """示例4: 获取期货数据"""
    print("\n" + "="*80)
    print("示例4: 获取期货数据")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取期货实时行情
    df = tool.get_futures_realtime(use_cache=False)
    print(f"\n期货实时行情数量: {len(df)}")
    print(df.head(10))
    
    # 获取沪深300期货历史数据
    df_history = tool.get_futures_history(
        symbol="IF0",
        start_date="20240101",
        end_date="202401231",
        use_cache=True
    )
    if not df_history.empty:
        print(f"\n沪深300期货历史数据 (最近5天):")
        print(df_history.tail(5))


def example_fund():
    """示例5: 获取基金数据"""
    print("\n" + "="*80)
    print("示例5: 获取基金数据")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取基金列表
    df = tool.get_fund_list(use_cache=True)
    print(f"\n基金总数: {len(df)}")
    print(df.head(10))
    
    # 获取特定基金的历史净值
    df_history = tool.get_fund_history(
        fund_code="000001",
        indicator="单位净值走势",
        use_cache=True
    )
    if not df_history.empty:
        print(f"\n华夏成长基金历史净值 (最近10天):")
        print(df_history.tail(10))


def example_macro():
    """示例6: 获取宏观经济数据"""
    print("\n" + "="*80)
    print("示例6: 获取宏观经济数据")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取GDP数据
    df_gdp = tool.get_macro_gdp(use_cache=True)
    print(f"\nGDP数据:")
    print(df_gdp)
    
    # 获取CPI数据
    df_cpi = tool.get_macro_cpi(use_cache=True)
    print(f"\nCPI数据 (最近10期):")
    print(df_cpi.tail(10))
    
    # 获取PMI数据
    df_pmi = tool.get_macro_pmi(use_cache=True)
    print(f"\nPMI数据 (最近10期):")
    print(df_pmi.tail(10))


def example_index():
    """示例7: 获取指数数据"""
    print("\n" + "="*80)
    print("示例7: 获取指数数据")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取指数实时行情
    df = tool.get_index_realtime(index_type="all", use_cache=False)
    print(f"\n指数实时行情:")
    print(df)
    
    # 获取上证指数历史数据
    df_history = tool.get_index_history(
        symbol="000001",
        period="daily",
        start_date="20240101",
        end_date="20241231",
        use_cache=True
    )
    print(f"\n上证指数历史数据 (最近10天):")
    print(df_history.tail(10))


def example_cache():
    """示例8: 缓存管理"""
    print("\n" + "="*80)
    print("示例8: 缓存管理")
    print("="*80)
    
    tool = AkShareTool(use_cache=True, cache_expiry_hours=24)
    
    # 第一次请求（从网络获取）
    print("\n第一次请求（从网络获取）:")
    df1 = tool.get_stock_realtime(symbol="000001", use_cache=True)
    
    # 第二次请求（从缓存获取）
    print("\n第二次请求（从缓存获取）:")
    df2 = tool.get_stock_realtime(symbol="000001", use_cache=True)
    
    # 清除缓存
    print("\n清除缓存:")
    count = tool.clear_cache()
    print(f"已清除 {count} 个缓存文件")
    
    # 再次请求（从网络获取）
    print("\n清除后再次请求:")
    df3 = tool.get_stock_realtime(symbol="000001", use_cache=True)


def example_error_handling():
    """示例9: 错误处理"""
    print("\n" + "="*80)
    print("示例9: 错误处理")
    print("="*80)
    
    tool = AkShareTool()
    
    try:
        # 尝试获取不存在的股票
        df = tool.get_stock_realtime(symbol="999999")
        if df.empty:
            print("股票代码不存在，返回空数据")
    except Exception as e:
        print(f"发生错误: {e}")


def example_data_analysis():
    """示例10: 数据分析"""
    print("\n" + "="*80)
    print("示例10: 数据分析")
    print("="*80)
    
    tool = AkShareTool()
    
    # 获取股票历史数据
    df = tool.get_stock_history(
        symbol="000001",
        period="daily",
        start_date="20240101",
        end_date="20241231",
        adjust="qfq"
    )
    
    if df.empty:
        print("没有数据")
        return
    
    # 计算收益率
    df['涨跌幅'] = df['收盘'].pct_change() * 100
    
    # 计算移动平均线
    df['MA5'] = df['收盘'].rolling(window=5).mean()
    df['MA10'] = df['收盘'].rolling(window=10).mean()
    df['MA20'] = df['收盘'].rolling(window=20).mean()
    
    # 计算波动率
    df['波动率'] = df['涨跌幅'].rolling(window=20).std()
    
    # 统计信息
    print("\n统计信息:")
    print(f"交易天数: {len(df)}")
    print(f"平均涨跌幅: {df['涨跌幅'].mean():.2f}%")
    print(f"最大涨幅: {df['涨跌幅'].max():.2f}%")
    print(f"最大跌幅: {df['涨跌幅'].min():.2f}%")
    print(f"平均波动率: {df['波动率'].mean():.2f}%")
    
    print("\n最近10天数据:")
    print(df[['日期', '收盘', '涨跌幅', 'MA5', 'MA10', 'MA20', '波动率']].tail(10))


def main():
    """运行所有示例"""
    print("\n" + "="*80)
    print("AkShare Tool 使用示例")
    print("="*80)
    
    examples = [
        ("A股实时行情", example_stock_realtime),
        ("股票历史数据", example_stock_history),
        ("港股数据", example_hk_stock),
        ("期货数据", example_futures),
        ("基金数据", example_fund),
        ("宏观经济数据", example_macro),
        ("指数数据", example_index),
        ("缓存管理", example_cache),
        ("错误处理", example_error_handling),
        ("数据分析", example_data_analysis),
    ]
    
    print("\n可用示例:")
    for i, (name, _) in enumerate(examples, 1):
        print(f"  {i}. {name}")
    
    print("\n运行所有示例可能需要较长时间，请耐心等待...")
    input("\n按Enter键开始运行所有示例，或Ctrl+C取消...")
    
    try:
        for name, func in examples:
            try:
                func()
            except Exception as e:
                print(f"\n示例 '{name}' 运行出错: {e}")
                import traceback
                traceback.print_exc()
        
        print("\n" + "="*80)
        print("所有示例运行完成！")
        print("="*80)
    
    except KeyboardInterrupt:
        print("\n\n用户中断执行")
    except Exception as e:
        print(f"\n发生错误: {e}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    main()

#!/bin/bash

# AkShare Installation Script
# This script installs AkShare and its dependencies

echo "Installing AkShare..."

# Check if pip is available
if ! command -v pip3 &> /dev/null; then
    echo "Error: pip3 not found. Please install Python and pip first."
    exit 1
fi

# Install AkShare
pip3 install akshare pandas numpy matplotlib

# Verify installation
python3 -c "import akshare; print(f'AkShare version: {akshare.__version__}')"

echo "AkShare installed successfully!"
echo ""
echo "Quick test:"
python3 -c "import akshare as ak; print('Testing AkShare...'); df = ak.stock_zh_a_spot_em(); print(f'Loaded {len(df)} stocks')"

echo ""
echo "For more information, see:"
echo "  - https://akshare.akfamily.xyz/"
echo "  - https://github.com/akfamily/akshare"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AkShare Tool 基本功能测试
验证核心功能是否正常工作
"""

import sys
from akshare_tool import AkShareTool


def test_imports():
    """测试导入"""
    print("测试1: 导入模块...")
    try:
        import akshare as ak
        import pandas as pd
        print("✓ 导入成功")
        return True
    except ImportError as e:
        print(f"✗ 导入失败: {e}")
        return False


def test_tool_initialization():
    """测试工具初始化"""
    print("\n测试2: 初始化工具...")
    try:
        tool = AkShareTool(use_cache=True, cache_expiry_hours=24)
        print("✓ 工具初始化成功")
        return True
    except Exception as e:
        print(f"✗ 初始化失败: {e}")
        return False


def test_cache_manager():
    """测试缓存管理器"""
    print("\n测试3: 缓存管理器...")
    try:
        tool = AkShareTool(use_cache=True)
        print("✓ 缓存管理器初始化成功")
        
        # 测试清除缓存
        count = tool.clear_cache()
        print(f"✓ 清除缓存成功，删除了 {count} 个文件")
        return True
    except Exception as e:
        print(f"✗ 缓存管理器测试失败: {e}")
        return False


def test_stock_realtime():
    """测试A股实时行情"""
    print("\n测试4: A股实时行情...")
    try:
        tool = AkShareTool(use_cache=False)
        df = tool.get_stock_realtime(symbol="000001", use_cache=False)
        
        if df is not None and not df.empty:
            print(f"✓ 获取成功，返回 {len(df)} 条记录")
            print(f"  列名: {list(df.columns)}")
            return True
        else:
            print("✗ 返回数据为空")
            return False
    except Exception as e:
        print(f"✗ 获取失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_macro_data():
    """测试宏观经济数据"""
    print("\n测试5: 宏观经济数据...")
    try:
        tool = AkShareTool(use_cache=True)
        
        # 测试GDP数据
        df_gdp = tool.get_macro_gdp(use_cache=True)
        if df_gdp is not None and not df_gdp.empty:
            print(f"✓ GDP数据获取成功，返回 {len(df_gdp)} 条记录")
        else:
            print("⚠ GDP数据为空")
        
        # 测试CPI数据
        df_cpi = tool.get_macro_cpi(use_cache=True)
        if df_cpi is not None and not df_cpi.empty:
            print(f"✓ CPI数据获取成功，返回 {len(df_cpi)} 条记录")
        else:
            print("⚠ CPI数据为空")
        
        # 测试PMI数据
        df_pmi = tool.get_macro_pmi(use_cache=True)
        if df_pmi is not None and not df_pmi.empty:
            print(f"✓ PMI数据获取成功，返回 {len(df_pmi)} 条记录")
        else:
            print("⚠ PMI数据为空")
        
        return True
    except Exception as e:
        print(f"✗ 获取失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_retry_mechanism():
    """测试重试机制"""
    print("\n测试6: 重试机制...")
    print("  (重试机制在网络错误时自动触发，此处仅验证装饰器正常工作)")
    try:
        tool = AkShareTool()
        # 重试机制会在实际网络请求时自动触发
        print("✓ 重试机制装饰器已加载")
        return True
    except Exception as e:
        print(f"✗ 重试机制测试失败: {e}")
        return False


def run_all_tests():
    """运行所有测试"""
    print("="*80)
    print("AkShare Tool 基本功能测试")
    print("="*80)
    
    tests = [
        test_imports,
        test_tool_initialization,
        test_cache_manager,
        test_stock_realtime,
        test_macro_data,
        test_retry_mechanism,
    ]
    
    results = []
    for test in tests:
        try:
            result = test()
            results.append(result)
        except Exception as e:
            print(f"\n✗ 测试异常: {e}")
            import traceback
            traceback.print_exc()
            results.append(False)
    
    # 统计结果
    print("\n" + "="*80)
    print("测试结果汇总")
    print("="*80)
    passed = sum(results)
    total = len(results)
    print(f"通过: {passed}/{total}")
    print(f"失败: {total - passed}/{total}")
    
    if passed == total:
        print("\n✓ 所有测试通过！")
        return 0
    else:
        print(f"\n✗ {total - passed} 个测试失败")
        return 1


if __name__ == "__main__":
    try:
        exit_code = run_all_tests()
        sys.exit(exit_code)
    except KeyboardInterrupt:
        print("\n\n测试被用户中断")
        sys.exit(1)
    except Exception as e:
        print(f"\n发生未预期的错误: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AkShare Tool 快速测试
只测试基本功能，不进行网络请求
"""

import sys


def test_imports():
    """测试导入"""
    print("测试1: 导入模块...")
    try:
        import akshare as ak
        import pandas as pd
        from akshare_tool import AkShareTool, CacheManager, retry_on_failure
        print("✓ 所有模块导入成功")
        print(f"  - AkShare version: {ak.__version__}")
        print(f"  - Pandas version: {pd.__version__}")
        return True
    except ImportError as e:
        print(f"✗ 导入失败: {e}")
        return False


def test_cache_manager():
    """测试缓存管理器"""
    print("\n测试2: 缓存管理器...")
    try:
        from akshare_tool import CacheManager
        import tempfile
        import os

        # 使用临时目录
        with tempfile.TemporaryDirectory() as tmpdir:
            cache = CacheManager(cache_dir=tmpdir, cache_expiry_hours=24)

            # 测试缓存键生成
            key1 = cache._get_cache_key("test_func", param1="value1", param2="value2")
            key2 = cache._get_cache_key("test_func", param1="value1", param2="value2")
            key3 = cache._get_cache_key("test_func", param1="value1", param2="value3")

            if key1 == key2:
                print("✓ 相同参数生成相同的缓存键")
            else:
                print("✗ 缓存键生成错误")
                return False

            if key1 != key3:
                print("✓ 不同参数生成不同的缓存键")
            else:
                print("✗ 缓存键生成错误")
                return False

            # 测试缓存读写
            import pandas as pd
            test_data = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})

            cache.set("test_func", test_data, param1="value1")
            cached_data = cache.get("test_func", param1="value1")

            if cached_data is not None and not cached_data.empty:
                print("✓ 缓存写入和读取成功")
            else:
                print("✗ 缓存读写失败")
                return False

            # 测试缓存清除
            count = cache.clear()
            print(f"✓ 清除缓存成功，删除了 {count} 个文件")

        return True
    except Exception as e:
        print(f"✗ 缓存管理器测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_tool_initialization():
    """测试工具初始化"""
    print("\n测试3: 工具初始化...")
    try:
        from akshare_tool import AkShareTool

        # 测试带缓存初始化
        tool1 = AkShareTool(use_cache=True, cache_expiry_hours=24)
        print("✓ 带缓存初始化成功")

        # 测试不带缓存初始化
        tool2 = AkShareTool(use_cache=False)
        print("✓ 不带缓存初始化成功")

        # 测试清除缓存
        count = tool1.clear_cache()
        print(f"✓ 清除缓存成功")

        return True
    except Exception as e:
        print(f"✗ 工具初始化失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_retry_decorator():
    """测试重试装饰器"""
    print("\n测试4: 重试装饰器...")
    try:
        from akshare_tool import retry_on_failure

        # 测试成功情况
        @retry_on_failure(max_retries=3, delay=0.1)
        def success_func():
            return "success"

        result = success_func()
        if result == "success":
            print("✓ 成功函数执行正常")
        else:
            print("✗ 成功函数返回错误")
            return False

        # 测试失败情况
        @retry_on_failure(max_retries=2, delay=0.1)
        def fail_func():
            raise ValueError("Test error")

        try:
            fail_func()
            print("✗ 失败函数应该抛出异常")
            return False
        except ValueError as e:
            if str(e) == "Test error":
                print("✓ 失败函数正确抛出异常")
            else:
                print("✗ 失败函数抛出错误的异常")
                return False

        return True
    except Exception as e:
        print(f"✗ 重试装饰器测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_command_line_help():
    """测试命令行帮助"""
    print("\n测试5: 命令行帮助...")
    try:
        import subprocess
        import os

        script_path = os.path.join(os.path.dirname(__file__), "akshare_tool.py")

        # 测试主帮助
        result = subprocess.run(
            ["python3", script_path, "--help"],
            capture_output=True,
            text=True,
            timeout=5
        )

        if result.returncode == 0 and "AkShare 金融市场数据获取工具" in result.stdout:
            print("✓ 主帮助信息显示正常")
        else:
            print("✗ 主帮助信息显示失败")
            return False

        # 测试子命令帮助
        result = subprocess.run(
            ["python3", script_path, "stock-realtime", "--help"],
            capture_output=True,
            text=True,
            timeout=5
        )

        if result.returncode == 0 and ("股票代码" in result.stdout or "symbol" in result.stdout):
            print("✓ 子命令帮助信息显示正常")
        else:
            print("✗ 子命令帮助信息显示失败")
            print(f"  退出码: {result.returncode}")
            print(f"  输出: {result.stdout[:200]}")
            return False

        return True
    except Exception as e:
        print(f"✗ 命令行帮助测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_api_structure():
    """测试API结构"""
    print("\n测试6: API结构...")
    try:
        from akshare_tool import AkShareTool

        tool = AkShareTool(use_cache=False)

        # 检查所有必需的方法是否存在
        required_methods = [
            'get_stock_realtime',
            'get_stock_history',
            'get_hk_stock_realtime',
            'get_hk_stock_history',
            'get_us_stock_realtime',
            'get_futures_realtime',
            'get_futures_history',
            'get_fund_list',
            'get_fund_history',
            'get_macro_gdp',
            'get_macro_cpi',
            'get_macro_pmi',
            'get_index_realtime',
            'get_index_history',
            'clear_cache',
        ]

        all_exist = True
        for method_name in required_methods:
            if hasattr(tool, method_name):
                print(f"  ✓ {method_name}")
            else:
                print(f"  ✗ {method_name} - 不存在")
                all_exist = False

        if all_exist:
            print("✓ 所有API方法都存在")
        else:
            print("✗ 部分API方法不存在")
            return False

        return True
    except Exception as e:
        print(f"✗ API结构测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def run_all_tests():
    """运行所有测试"""
    print("="*80)
    print("AkShare Tool 快速功能测试")
    print("="*80)
    print("注意: 此测试不进行实际网络请求，仅验证代码结构和基本功能\n")

    tests = [
        test_imports,
        test_cache_manager,
        test_tool_initialization,
        test_retry_decorator,
        test_command_line_help,
        test_api_structure,
    ]

    results = []
    for test in tests:
        try:
            result = test()
            results.append(result)
        except Exception as e:
            print(f"\n✗ 测试异常: {e}")
            import traceback
            traceback.print_exc()
            results.append(False)

    # 统计结果
    print("\n" + "="*80)
    print("测试结果汇总")
    print("="*80)
    passed = sum(results)
    total = len(results)
    print(f"通过: {passed}/{total}")
    print(f"失败: {total - passed}/{total}")

    if passed == total:
        print("\n✓ 所有测试通过！")
        print("\n提示: 如需测试实际数据获取功能，请运行:")
        print("  python3 test_basic.py")
        return 0
    else:
        print(f"\n✗ {total - passed} 个测试失败")
        return 1


if __name__ == "__main__":
    try:
        exit_code = run_all_tests()
        sys.exit(exit_code)
    except KeyboardInterrupt:
        print("\n\n测试被用户中断")
        sys.exit(1)
    except Exception as e:
        print(f"\n发生未预期的错误: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)