Akshare

#!/usr/bin/env python3
"""AKShare 债券数据获取脚本
支持可转债、债券报价等数据"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_bond_convertible():
    """获取可转债列表"""
    try:
        df = ak.bond_zh_cov()
        return {
            "type": "bond_convertible",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取可转债列表失败: {str(e)}"}

def get_bond_convertible_hist(symbol="sz123456"):
    """获取可转债历史K线数据"""
    try:
        df = ak.bond_zh_hs_cov_daily(symbol=symbol)
        return {
            "type": "bond_convertible_hist",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取可转债历史数据失败: {str(e)}"}

def get_bond_spot_quote():
    """获取中国债券现货报价"""
    try:
        df = ak.bond_spot_quote()
        return {
            "type": "bond_spot_quote",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取债券现货报价失败: {str(e)}"}

def get_bond_cov_jsl():
    """获取集思录可转债数据"""
    try:
        df = ak.bond_cov_jsl()
        return {
            "type": "bond_cov_jsl",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取集思录可转债数据失败: {str(e)}"}

def get_bond_zh_hs_cov():
    """获取沪深可转债数据"""
    try:
        df = ak.bond_zh_hs_cov()
        return {
            "type": "bond_zh_hs_cov",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取沪深可转债数据失败: {str(e)}"}

def get_bond_treasure_issue():
    """获取国债发行数据"""
    try:
        df = ak.bond_treasure_issue()
        return {
            "type": "bond_treasure_issue",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取国债发行数据失败: {str(e)}"}

def get_bond_local_government_issue():
    """获取地方债发行数据"""
    try:
        df = ak.bond_local_government_issue()
        return {
            "type": "bond_local_government_issue",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取地方债发行数据失败: {str(e)}"}

def get_bond_corporate_issue():
    """获取企业债发行数据"""
    try:
        df = ak.bond_corporate_issue()
        return {
            "type": "bond_corporate_issue",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取企业债发行数据失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 债券数据获取')
    parser.add_argument('--type', required=True,
                       choices=['convertible', 'convertible_hist', 'spot_quote', 'cov_jsl',
                               'zh_hs_cov', 'treasure_issue', 'local_gov_issue', 'corporate_issue'],
                       help='数据类型')
    parser.add_argument('--symbol', default='sz123456', help='债券代码')

    args = parser.parse_args()

    if args.type == 'convertible':
        result = get_bond_convertible()
    elif args.type == 'convertible_hist':
        result = get_bond_convertible_hist(args.symbol)
    elif args.type == 'spot_quote':
        result = get_bond_spot_quote()
    elif args.type == 'cov_jsl':
        result = get_bond_cov_jsl()
    elif args.type == 'zh_hs_cov':
        result = get_bond_zh_hs_cov()
    elif args.type == 'treasure_issue':
        result = get_bond_treasure_issue()
    elif args.type == 'local_gov_issue':
        result = get_bond_local_government_issue()
    elif args.type == 'corporate_issue':
        result = get_bond_corporate_issue()
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 加密货币数据获取脚本
支持加密货币价格、K线数据等"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_crypto_symbols():
    """获取加密货币交易对列表"""
    try:
        df = ak.crypto_binance_symbols()
        return {
            "type": "crypto_symbols",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取加密货币交易对失败: {str(e)}"}

def get_crypto_price(symbol="BTC/USDT"):
    """获取加密货币实时价格"""
    try:
        symbol_map = {
            "BTC/USDT": ak.crypto_binance_btc_usdt_spot,
            "ETH/USDT": ak.crypto_binance_eth_usdt_spot,
            "BNB/USDT": ak.crypto_binance_bnb_usdt_spot,
        }

        func = symbol_map.get(symbol, ak.crypto_binance_btc_usdt_spot)
        df = func()

        if df is not None and len(df) > 0:
            latest = df.iloc[-1]
            return {
                "type": "crypto_price",
                "symbol": symbol,
                "price": float(latest.get('close', latest.get('price', 0))),
                "volume": str(latest.get('volume', 'N/A')),
                "timestamp": str(latest.get('date', pd.Timestamp.now()))
            }
        return {"error": "No data available"}
    except Exception as e:
        return {"error": f"获取加密货币价格失败: {str(e)}"}

def get_crypto_kline(symbol="BTC/USDT", period="daily"):
    """获取加密货币K线数据"""
    try:
        if symbol.upper() in ["BTC/USDT", "BTC", "btc"]:
            df = ak.crypto_binance_btc_usdt_kline(period=period)
        elif symbol.upper() in ["ETH/USDT", "ETH", "eth"]:
            df = ak.crypto_binance_eth_usdt_kline(period=period)
        else:
            return {"error": f"不支持的交易对: {symbol}"}

        return {
            "type": "crypto_kline",
            "symbol": symbol,
            "period": period,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取加密货币K线失败: {str(e)}"}

def get_crypto_all_spot():
    """获取所有加密货币实时价格"""
    try:
        # 获取币安所有交易对
        df = ak.crypto_binance_symbols()
        symbols = df['symbol'].tolist()[:10]  # 限制前10个避免请求过多

        results = []
        for symbol in symbols:
            try:
                price_data = get_crypto_price(f"{symbol}/USDT")
                if "error" not in price_data:
                    results.append(price_data)
            except:
                continue

        return {
            "type": "crypto_all_spot",
            "data": results,
            "count": len(results)
        }
    except Exception as e:
        return {"error": f"获取所有加密货币价格失败: {str(e)}"}

def get_crypto_market_cap():
    """获取加密货币市值信息"""
    try:
        df = ak.crypto_bitcoin_market_cap()
        return {
            "type": "crypto_market_cap",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取加密货币市值失败: {str(e)}"}

def get_crypto_global_index():
    """获取加密货币全球指数"""
    try:
        df = ak.crypto_bitcoin_global_index()
        return {
            "type": "crypto_global_index",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取加密货币全球指数失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 加密货币数据获取')
    parser.add_argument('--type', required=True,
                       choices=['symbols', 'price', 'kline', 'all_spot', 'market_cap', 'global_index'],
                       help='数据类型')
    parser.add_argument('--symbol', default='BTC/USDT', help='加密货币交易对')
    parser.add_argument('--period', default='daily',
                       choices=['daily', 'weekly', 'monthly'],
                       help='K线周期')

    args = parser.parse_args()

    if args.type == 'symbols':
        result = get_crypto_symbols()
    elif args.type == 'price':
        result = get_crypto_price(args.symbol)
    elif args.type == 'kline':
        result = get_crypto_kline(args.symbol, args.period)
    elif args.type == 'all_spot':
        result = get_crypto_all_spot()
    elif args.type == 'market_cap':
        result = get_crypto_market_cap()
    elif args.type == 'global_index':
        result = get_crypto_global_index()
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 外汇贵金属数据获取脚本
支持外汇行情、汇率数据、贵金属价格等"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_forex_spot():
    """获取外汇实时行情"""
    try:
        df = ak.forex_spot_em()
        return {
            "type": "forex_spot",
            "data": df.to_dict('records'),
            "count": len(df),
            "update_time": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
        }
    except Exception as e:
        return {"error": f"获取外汇实时行情失败: {str(e)}"}

def get_fx_spot_quote():
    """获取外汇即期报价"""
    try:
        df = ak.fx_spot_quote()
        return {
            "type": "fx_spot_quote",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取外汇即期报价失败: {str(e)}"}

def get_fx_swap_quote():
    """获取外汇掉期报价"""
    try:
        df = ak.fx_swap_quote()
        return {
            "type": "fx_swap_quote",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取外汇掉期报价失败: {str(e)}"}

def get_usd_cny():
    """获取美元兑人民币汇率"""
    try:
        df = ak.forex_usd_cny()
        return {
            "type": "usd_cny",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取美元兑人民币汇率失败: {str(e)}"}

def get_metals_gold():
    """获取国际金价"""
    try:
        df = ak.metals_gold()
        return {
            "type": "metals_gold",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取国际金价失败: {str(e)}"}

def get_metals_silver():
    """获取国际银价"""
    try:
        df = ak.metals_silver()
        return {
            "type": "metals_silver",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取国际银价失败: {str(e)}"}

def get_metals_shibor():
    """获取上海银行间拆借利率"""
    try:
        df = ak.metals_shibor()
        return {
            "type": "metals_shibor",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取上海银行间拆借利率失败: {str(e)}"}

def get_currency_pair_rate(pair="USD/CNY"):
    """获取指定货币对汇率"""
    try:
        # 根据货币对选择对应接口
        if pair.upper() in ["USD/CNY", "USDCNY"]:
            df = ak.forex_usd_cny()
        elif pair.upper() in ["EUR/CNY", "EURCNY"]:
            df = ak.forex_eur_cny()
        elif pair.upper() in ["GBP/CNY", "GBPCNY"]:
            df = ak.forex_gbp_cny()
        elif pair.upper() in ["JPY/CNY", "JPYCNY"]:
            df = ak.forex_jpy_cny()
        else:
            # 尝试通用外汇接口
            df = ak.forex_hist(symbol=pair.replace("/", ""))

        return {
            "type": "currency_pair_rate",
            "pair": pair,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取货币对汇率失败: {str(e)}"}

def get_forex_trading_calendar():
    """获取外汇交易日历"""
    try:
        df = ak.fx_trading_calendar()
        return {
            "type": "forex_trading_calendar",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取外汇交易日历失败: {str(e)}"}

def get_currency_rates_boc():
    """获取中国银行外汇牌价"""
    try:
        df = ak.currency_boc_sina()
        return {
            "type": "currency_rates_boc",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取中国银行外汇牌价失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 外汇贵金属数据获取')
    parser.add_argument('--type', required=True,
                       choices=['spot', 'spot_quote', 'swap_quote', 'usd_cny', 'gold',
                               'silver', 'shibor', 'pair_rate', 'calendar', 'boc_rates'],
                       help='数据类型')
    parser.add_argument('--symbol', help='货币对或贵金属代码')
    parser.add_argument('--pair', default='USD/CNY', help='货币对，如 USD/CNY')

    args = parser.parse_args()

    if args.type == 'spot':
        result = get_forex_spot()
    elif args.type == 'spot_quote':
        result = get_fx_spot_quote()
    elif args.type == 'swap_quote':
        result = get_fx_swap_quote()
    elif args.type == 'usd_cny':
        result = get_usd_cny()
    elif args.type == 'gold':
        result = get_metals_gold()
    elif args.type == 'silver':
        result = get_metals_silver()
    elif args.type == 'shibor':
        result = get_metals_shibor()
    elif args.type == 'pair_rate':
        pair = args.pair if args.pair else "USD/CNY"
        result = get_currency_pair_rate(pair)
    elif args.type == 'calendar':
        result = get_forex_trading_calendar()
    elif args.type == 'boc_rates':
        result = get_currency_rates_boc()
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 基金数据获取脚本
支持ETF、开放式基金、基金评级等数据"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_etf_spot():
    """获取ETF实时行情"""
    try:
        df = ak.fund_etf_spot_em()
        return {
            "type": "etf_spot",
            "data": df.to_dict('records'),
            "count": len(df),
            "update_time": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
        }
    except Exception as e:
        return {"error": f"获取ETF实时行情失败: {str(e)}"}

def get_etf_hist(symbol, period="daily", start_date=None, end_date=None, adjust="qfq"):
    """获取ETF历史K线数据"""
    try:
        df = ak.fund_etf_hist_em(
            symbol=symbol,
            period=period,
            start_date=start_date,
            end_date=end_date,
            adjust=adjust
        )
        return {
            "type": "etf_hist",
            "symbol": symbol,
            "period": period,
            "adjust": adjust,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取ETF历史数据失败: {str(e)}"}

def get_open_fund_daily(symbol):
    """获取开放式基金每日净值"""
    try:
        df = ak.fund_open_fund_daily_em(symbol=symbol)
        return {
            "type": "open_fund_daily",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取开放式基金净值失败: {str(e)}"}

def get_fund_rating():
    """获取基金评级"""
    try:
        df = ak.fund_rating_all()
        return {
            "type": "fund_rating",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取基金评级失败: {str(e)}"}

def get_fund_portfolio(symbol):
    """获取基金持仓信息"""
    try:
        df = ak.fund_portfolio_hold_em(symbol=symbol)
        return {
            "type": "fund_portfolio",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取基金持仓失败: {str(e)}"}

def get_fund_manager():
    """获取基金经理信息"""
    try:
        df = ak.fund_manager()
        return {
            "type": "fund_manager",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取基金经理信息失败: {str(e)}"}

def get_money_fund():
    """获取货币基金信息"""
    try:
        df = ak.fund_money_fund_em()
        return {
            "type": "money_fund",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取货币基金信息失败: {str(e)}"}

def get_fund_scale():
    """获取基金规模信息"""
    try:
        df = ak.fund_scale_open_em()
        return {
            "type": "fund_scale",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取基金规模信息失败: {str(e)}"}

def get_fund_new():
    """获取新发基金信息"""
    try:
        df = ak.fund_new_found_em()
        return {
            "type": "fund_new",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取新发基金信息失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 基金数据获取')
    parser.add_argument('--type', required=True,
                       choices=['etf_spot', 'etf_hist', 'open_fund', 'rating', 'portfolio',
                               'manager', 'money_fund', 'scale', 'new'],
                       help='数据类型')
    parser.add_argument('--symbol', help='基金代码')
    parser.add_argument('--period', default='daily',
                       choices=['daily', 'weekly', 'monthly'],
                       help='周期')
    parser.add_argument('--start', help='开始日期 YYYYMMDD')
    parser.add_argument('--end', help='结束日期 YYYYMMDD')
    parser.add_argument('--adjust', default='qfq',
                       choices=['', 'qfq', 'hfq'],
                       help='复权类型')

    args = parser.parse_args()

    if args.type == 'etf_spot':
        result = get_etf_spot()
    elif args.type == 'etf_hist':
        if not args.symbol:
            result = {"error": "ETF历史数据需要指定symbol参数"}
        else:
            result = get_etf_hist(args.symbol, args.period, args.start, args.end, args.adjust)
    elif args.type == 'open_fund':
        if not args.symbol:
            result = {"error": "开放式基金数据需要指定symbol参数"}
        else:
            result = get_open_fund_daily(args.symbol)
    elif args.type == 'rating':
        result = get_fund_rating()
    elif args.type == 'portfolio':
        if not args.symbol:
            result = {"error": "基金持仓数据需要指定symbol参数"}
        else:
            result = get_fund_portfolio(args.symbol)
    elif args.type == 'manager':
        result = get_fund_manager()
    elif args.type == 'money_fund':
        result = get_money_fund()
    elif args.type == 'scale':
        result = get_fund_scale()
    elif args.type == 'new':
        result = get_fund_new()
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 期货数据获取脚本
支持期货行情、历史K线、库存数据等"""
import sys
import json
import argparse
import pandas as pd
from akshare import get_futures_daily
import akshare as ak

def get_futures_daily_data(start_date, end_date, market="CFFEX"):
    """获取期货日线数据"""
    try:
        df = get_futures_daily(
            start_date=start_date,
            end_date=end_date,
            market=market
        )
        return {
            "market": market,
            "start_date": start_date,
            "end_date": end_date,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取期货日线数据失败: {str(e)}"}

def get_futures_spot():
    """获取期货实时行情"""
    try:
        df = ak.futures_zh_spot()
        return {
            "data": df.to_dict('records'),
            "count": len(df),
            "update_time": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
        }
    except Exception as e:
        return {"error": f"获取期货实时行情失败: {str(e)}"}

def get_futures_inventory(symbol="豆一"):
    """获取期货库存数据"""
    try:
        df = ak.futures_inventory_99(symbol=symbol)
        return {
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取期货库存数据失败: {str(e)}"}

def get_cffex_daily(start_date, end_date):
    """获取中金所期货数据"""
    return get_futures_daily_data(start_date, end_date, "CFFEX")

def get_shfe_daily(start_date, end_date):
    """获取上期所期货数据"""
    return get_futures_daily_data(start_date, end_date, "SHFE")

def get_dce_daily(start_date, end_date):
    """获取大商所期货数据"""
    return get_futures_daily_data(start_date, end_date, "DCE")

def get_czce_daily(start_date, end_date):
    """获取郑商所期货数据"""
    return get_futures_daily_data(start_date, end_date, "CZCE")

def get_ine_daily(start_date, end_date):
    """获取上海国际能源交易中心期货数据"""
    return get_futures_daily_data(start_date, end_date, "INE")

def get_gfex_daily(start_date, end_date):
    """获取广期所期货数据"""
    return get_futures_daily_data(start_date, end_date, "GFEX")

def get_futures_contracts(exchange="CFFEX"):
    """获取期货合约列表"""
    try:
        # 获取实时行情来提取合约信息
        df = ak.futures_zh_spot()
        # 根据交易所筛选
        if exchange == "CFFEX":
            contracts = df[df['symbol'].str.contains('IF|IC|IH|T|TF|TS')]
        elif exchange == "SHFE":
            contracts = df[df['symbol'].str.contains('AU|AG|CU|AL|ZN|PB|NI|SN|RB|WR|HC')]
        elif exchange == "DCE":
            contracts = df[df['symbol'].str.contains('C|CS|A|M|Y|B|P|FB|BB|L|V|PP')]
        elif exchange == "CZCE":
            contracts = df[df['symbol'].str.contains('SR|CF|TA|OI|RI|WH|RS|RM|JR|LR|SF|SM')]
        else:
            contracts = df

        return {
            "exchange": exchange,
            "contracts": contracts[['symbol', 'contract']].drop_duplicates().to_dict('records'),
            "count": len(contracts)
        }
    except Exception as e:
        return {"error": f"获取期货合约列表失败: {str(e)}"}

def get_futures_main_contract(symbol="IF"):
    """获取期货主力合约"""
    try:
        df = ak.futures_main_sure_em(symbol=symbol)
        return {
            "symbol": symbol,
            "main_contract": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取期货主力合约失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 期货数据获取')
    parser.add_argument('--type', required=True,
                       choices=['daily', 'spot', 'inventory', 'contracts', 'main_contract',
                               'cffex', 'shfe', 'dce', 'czce', 'ine', 'gfex'],
                       help='数据类型')
    parser.add_argument('--start', required=True, help='开始日期 YYYYMMDD')
    parser.add_argument('--end', required=True, help='结束日期 YYYYMMDD')
    parser.add_argument('--market', default='CFFEX',
                       choices=['CFFEX', 'SHFE', 'DCE', 'CZCE', 'INE', 'GFEX'],
                       help='交易所')
    parser.add_argument('--symbol', help='期货品种或合约代码')

    args = parser.parse_args()

    if args.type == 'daily':
        result = get_futures_daily_data(args.start, args.end, args.market)
    elif args.type == 'spot':
        result = get_futures_spot()
    elif args.type == 'inventory':
        symbol = args.symbol if args.symbol else "豆一"
        result = get_futures_inventory(symbol)
    elif args.type == 'contracts':
        market = args.market if args.market else "CFFEX"
        result = get_futures_contracts(market)
    elif args.type == 'main_contract':
        symbol = args.symbol if args.symbol else "IF"
        result = get_futures_main_contract(symbol)
    elif args.type == 'cffex':
        result = get_cffex_daily(args.start, args.end)
    elif args.type == 'shfe':
        result = get_shfe_daily(args.start, args.end)
    elif args.type == 'dce':
        result = get_dce_daily(args.start, args.end)
    elif args.type == 'czce':
        result = get_czce_daily(args.start, args.end)
    elif args.type == 'ine':
        result = get_ine_daily(args.start, args.end)
    elif args.type == 'gfex':
        result = get_gfex_daily(args.start, args.end)
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 指数数据获取脚本
支持A股指数、全球指数等数据"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_index_daily(symbol="sh000001"):
    """获取A股指数日线数据"""
    try:
        df = ak.stock_zh_index_daily_em(symbol=symbol)
        return {
            "type": "index_daily",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取指数日线数据失败: {str(e)}"}

def get_index_component(symbol="000300"):
    """获取指数成分股"""
    try:
        df = ak.index_stock_cons_csindex(symbol=symbol)
        return {
            "type": "index_component",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取指数成分股失败: {str(e)}"}

def get_global_index():
    """获取全球指数数据"""
    try:
        df = ak.index_global_em()
        return {
            "type": "global_index",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as as e:
        return {"error": f"获取全球指数失败: {str(e)}"}

def get_index_value(symbol="000300"):
    """获取指数估值数据"""
    try:
        df = ak.index_value_name_funddb()
        # 筛选指定指数
        if symbol:
            df = df[df['指数代码'] == symbol]
        return {
            "type": "index_value",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取指数估值失败: {str(e)}"}

def get_index_weight(symbol="000300"):
    """获取指数权重数据"""
    try:
        df = ak.index_stock_weight_csindex(symbol=symbol)
        return {
            "type": "index_weight",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取指数权重失败: {str(e)}"}

def get_index_analysis(symbol="000300"):
    """获取指数分析数据"""
    try:
        df = ak.index_analysis_weekly(symbol=symbol)
        return {
            "type": "index_analysis",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取指数分析失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 指数数据获取')
    parser.add_argument('--type', required=True,
                       choices=['daily', 'component', 'global', 'value', 'weight', 'analysis'],
                       help='数据类型')
    parser.add_argument('--symbol', default='000300', help='指数代码')

    args = parser.parse_args()

    if args.type == 'daily':
        # 处理带前缀的指数代码
        symbol = args.symbol
        if not symbol.startswith(('sh', 'sz')):
            if symbol.startswith('000'):
                symbol = f"sh{symbol}"
            elif symbol.startswith('399'):
                symbol = f"sz{symbol}"
        result = get_index_daily(symbol)
    elif args.type == 'component':
        result = get_index_component(args.symbol)
    elif args.type == 'global':
        result = get_global_index()
    elif args.type == 'value':
        result = get_index_value(args.symbol)
    elif args.type == 'weight':
        result = get_index_weight(args.symbol)
    elif args.type == 'analysis':
        result = get_index_analysis(args.symbol)
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 宏观经济数据获取脚本
支持GDP、CPI、PMI等宏观经济指标"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_macro_china_gdp():
    """获取中国GDP数据"""
    try:
        df = ak.macro_china_gdp()
        return {
            "type": "macro_china_gdp",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取中国GDP数据失败: {str(e)}"}

def get_macro_china_gdp_yearly():
    """获取中国GDP年度数据"""
    try:
        df = ak.macro_china_gdp_yearly()
        return {
            "type": "macro_china_gdp_yearly",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取中国GDP年度数据失败: {str(e)}"}

def get_macro_china_cpi():
    """获取中国CPI数据"""
    try:
        df = ak.macro_china_cpi()
        return {
            "type": "macro_china_cpi",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取中国CPI数据失败: {str(e)}"}

def get_macro_china_cpi_yearly():
    """获取中国CPI年度数据"""
    try:
        df = ak.macro_china_cpi_yearly()
        return {
            "type": "macro_china_cpi_yearly",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取中国CPI年度数据失败: {str(e)}"}

def get_macro_china_pmi():
    """获取中国PMI数据"""
    try:
        df = ak.macro_china_pmi()
        return {
            "type": "macro_china_pmi",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取中国PMI数据失败: {str(e)}"}

def get_macro_china_m2():
    """获取中国M2货币供应数据"""
    try:
        df = ak.macro_china_m2()
        return {
            "type": "macro_china_m2",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取中国M2数据失败: {str(e)}"}

def get_macro_usa_non_farm():
    """获取美国非农就业数据"""
    try:
        df = ak.macro_usa_non_farm()
        return {
            "type": "macro_usa_non_farm",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取美国非农就业数据失败: {str(e)}"}

def get_macro_usa_cpi_monthly():
    """获取美国CPI月度数据"""
    try:
        df = ak.macro_usa_cpi_monthly()
        return {
            "type": "macro_usa_cpi_monthly",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取美国CPI月度数据失败: {str(e)}"}

def get_macro_usa_gdp_monthly():
    """获取美国GDP月度数据"""
    try:
        df = ak.macro_usa_gdp_monthly()
        return {
            "type": "macro_usa_gdp_monthly",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取美国GDP月度数据失败: {str(e)}"}

def get_macro_usa_unemployment():
    """获取美国失业率数据"""
    try:
        df = ak.macro_usa_unemployment()
        return {
            "type": "macro_usa_unemployment",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取美国失业率数据失败: {str(e)}"}

def get_macro_summary():
    """获取宏观经济概览"""
    try:
        result = {}

        # 中国主要指标
        china_indicators = {
            "gdp": get_macro_china_gdp(),
            "cpi": get_macro_china_cpi(),
            "pmi": get_macro_china_pmi(),
            "m2": get_macro_china_m2()
        }

        # 美国主要指标
        usa_indicators = {
            "non_farm": get_macro_usa_non_farm(),
            "cpi": get_macro_usa_cpi_monthly(),
            "gdp": get_macro_usa_gdp_monthly(),
            "unemployment": get_macro_usa_unemployment()
        }

        # 提取最新数据
        for key, data in china_indicators.items():
            if "error" not in data and len(data.get("data", [])) > 0:
                latest = data["data"][-1]
                result[f"china_{key}"] = {
                    "value": latest.get("value", latest.get("data", None)),
                    "date": latest.get("date", latest.get("time", None))
                }

        for key, data in usa_indicators.items():
            if "error" not in data and len(data.get("data", [])) > 0:
                latest = data["data"][-1]
                result[f"usa_{key}"] = {
                    "value": latest.get("value", latest.get("data", None)),
                    "date": latest.get("date", latest.get("time", None))
                }

        return {
            "type": "macro_summary",
            "data": result,
            "update_time": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
        }
    except Exception as e:
        return {"error": f"获取宏观经济概览失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 宏观经济数据获取')
    parser.add_argument('--type', required=True,
                       choices=['gdp', 'gdp_yearly', 'cpi', 'cpi_yearly', 'pmi', 'm2',
                               'usa_non_farm', 'usa_cpi', 'usa_gdp', 'usa_unemployment', 'summary'],
                       help='宏观经济指标类型')

    args = parser.parse_args()

    if args.type == 'gdp':
        result = get_macro_china_gdp()
    elif args.type == 'gdp_yearly':
        result = get_macro_china_gdp_yearly()
    elif args.type == 'cpi':
        result = get_macro_china_cpi()
    elif args.type == 'cpi_yearly':
        result = get_macro_china_cpi_yearly()
    elif args.type == 'pmi':
        result = get_macro_china_pmi()
    elif args.type == 'm2':
        result = get_macro_china_m2()
    elif args.type == 'usa_non_farm':
        result = get_macro_usa_non_farm()
    elif args.type == 'usa_cpi':
        result = get_macro_usa_cpi_monthly()
    elif args.type == 'usa_gdp':
        result = get_macro_usa_gdp_monthly()
    elif args.type == 'usa_unemployment':
        result = get_macro_usa_unemployment()
    elif args.type == 'summary':
        result = get_macro_summary()
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 期权数据获取脚本
支持期权历史数据、实时行情等"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_option_hist_dce(symbol="豆粕期权"):
    """获取大商所期权历史数据"""
    try:
        df = ak.option_hist_dce(symbol=symbol)
        return {
            "type": "option_hist_dce",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取大商所期权历史数据失败: {str(e)}"}

def get_option_hist_czce(symbol="白糖期权"):
    """获取郑商所期权历史数据"""
    try:
        df = ak.option_hist_czce(symbol=symbol)
        return {
            "type": "option_hist_czce",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取郑商所期权历史数据失败: {str(e)}"}

def get_option_hist_shfe(symbol="铜期权"):
    """获取上期所期权历史数据"""
    try:
        df = ak.option_hist_shfe(symbol=symbol)
        return {
            "type": "option_hist_shfe",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取上期所期权历史数据失败: {str(e)}"}

def get_option_sse_spot_price(symbol="510050"):
    """获取上证50ETF期权实时行情"""
    try:
        df = ak.option_sse_spot_price(symbol=symbol)
        return {
            "type": "option_sse_spot",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取上证50ETF期权实时行情失败: {str(e)}"}

def get_option_sse_underlying_spot_price(symbol="510050"):
    """获取上证50ETF期权标的实时行情"""
    try:
        df = ak.option_sse_underlying_spot_price(symbol=symbol)
        return {
            "type": "option_sse_underlying_spot",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取期权标的实时行情失败: {str(e)}"}

def get_option_sse_minute_sina(symbol="510050"):
    """获取上证50ETF期权分钟数据"""
    try:
        df = ak.option_sse_minute_sina(symbol=symbol)
        return {
            "type": "option_sse_minute",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取期权分钟数据失败: {str(e)}"}

def get_option_sse_daily_sina(symbol="510050"):
    """获取上证50ETF期权日K线数据"""
    try:
        df = ak.option_sse_daily_sina(symbol=symbol)
        return {
            "type": "option_sse_daily",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取期权日K线数据失败: {str(e)}"}

def get_option_finance_board(symbol="510050"):
    """获取期权财务指标"""
    try:
        df = ak.option_finance_board(symbol=symbol)
        return {
            "type": "option_finance_board",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取期权财务指标失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 期权数据获取')
    parser.add_argument('--type', required=True,
                       choices=['hist_dce', 'hist_czce', 'hist_shfe', 'sse_spot',
                               'sse_underlying', 'sse_minute', 'sse_daily', 'finance_board'],
                       help='数据类型')
    parser.add_argument('--symbol', default='510050', help='期权代码或名称')

    args = parser.parse_args()

    if args.type == 'hist_dce':
        result = get_option_hist_dce(args.symbol)
    elif args.type == 'hist_czce':
        result = get_option_hist_czce(args.symbol)
    elif args.type == 'hist_shfe':
        result = get_option_hist_shfe(args.symbol)
    elif args.type == 'sse_spot':
        result = get_option_sse_spot_price(args.symbol)
    elif args.type == 'sse_underlying':
        result = get_option_sse_underlying_spot_price(args.symbol)
    elif args.type == 'sse_minute':
        result = get_option_sse_minute_sina(args.symbol)
    elif args.type == 'sse_daily':
        result = get_option_sse_daily_sina(args.symbol)
    elif args.type == 'finance_board':
        result = get_option_finance_board(args.symbol)
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AKShare 特色数据获取脚本
支持龙虎榜、融资融券、北向资金、股东数据等"""
import sys
import json
import argparse
import pandas as pd
import akshare as ak

def get_stock_lhb_detail(start_date, end_date):
    """获取龙虎榜详细数据"""
    try:
        df = ak.stock_lhb_detail_em(start_date=start_date, end_date=end_date)
        return {
            "type": "lhb_detail",
            "start_date": start_date,
            "end_date": end_date,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取龙虎榜详细数据失败: {str(e)}"}

def get_stock_lhb_hyyyb(start_date, end_date):
    """获取龙虎榜营业部排名"""
    try:
        df = ak.stock_lhb_hyyyb_em(start_date=start_date, end_date=end_date)
        return {
            "type": "lhb_hyyyb",
            "start_date": start_date,
            "end_date": end_date,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取龙虎榜营业部排名失败: {str(e)}"}

def get_stock_margin_sse(start_date, end_date):
    """获取沪深融资融券汇总数据"""
    try:
        df = ak.stock_margin_sse(start_date=start_date, end_date=end_date)
        return {
            "type": "margin_sse",
            "start_date": start_date,
            "end_date": end_date,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取融资融券汇总数据失败: {str(e)}"}

def get_stock_margin_detail(date):
    """获取个股融资融券明细"""
    try:
        df = ak.stock_margin_detail_sse(date=date)
        return {
            "type": "margin_detail",
            "date": date,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取融资融券明细失败: {str(e)}"}

def get_stock_hsgt_hist(symbol="北向资金"):
    """获取北向资金历史数据"""
    try:
        df = ak.stock_hsgt_hist_em(symbol=symbol)
        return {
            "type": "hsgt_hist",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取北向资金历史数据失败: {str(e)}"}

def get_stock_hsgt_hold_stock(market="北向", indicator="今日排行"):
    """获取北向资金持股明细"""
    try:
        df = ak.stock_hsgt_hold_stock_em(market=market, indicator=indicator)
        return {
            "type": "hsgt_hold_stock",
            "market": market,
            "indicator": indicator,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取北向资金持股明细失败: {str(e)}"}

def get_stock_gdfx_top_10(symbol, date):
    """获取前十大股东"""
    try:
        df = ak.stock_gdfx_top_10_em(symbol=symbol, date=date)
        return {
            "type": "gdfx_top_10",
            "symbol": symbol,
            "date": date,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取前十大股东失败: {str(e)}"}

def get_stock_gdfx_free_top_10(symbol, date):
    """获取前十大流通股东"""
    try:
        df = ak.stock_gdfx_free_top_10_em(symbol=symbol, date=date)
        return {
            "type": "gdfx_free_top_10",
            "symbol": symbol,
            "date": date,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取前十大流通股东失败: {str(e)}"}

def get_stock_board_industry():
    """获取行业板块行情"""
    try:
        df = ak.stock_board_industry_name_em()
        return {
            "type": "board_industry",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取行业板块行情失败: {str(e)}"}

def get_stock_board_concept():
    """获取概念板块行情"""
    try:
        df = ak.stock_board_concept_name_em()
        return {
            "type": "board_concept",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取概念板块行情失败: {str(e)}"}

def get_stock_board_industry_cons(symbol="银行"):
    """获取特定板块成分股"""
    try:
        df = ak.stock_board_industry_cons_em(symbol=symbol)
        return {
            "type": "board_industry_cons",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取板块成分股失败: {str(e)}"}

def get_stock_restricted_release(symbol="全部A股"):
    """获取限售解禁数据"""
    try:
        df = ak.stock_restricted_release_queue_em(symbol=symbol)
        return {
            "type": "restricted_release",
            "symbol": symbol,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取限售解禁数据失败: {str(e)}"}

def get_stock_market_fund_flow():
    """获取大盘资金流向"""
    try:
        df = ak.stock_market_fund_flow()
        return {
            "type": "market_fund_flow",
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取大盘资金流向失败: {str(e)}"}

def get_stock_individual_fund_flow(stock, market="sz"):
    """获取个股资金流向"""
    try:
        df = ak.stock_individual_fund_flow(stock=stock, market=market)
        return {
            "type": "individual_fund_flow",
            "stock": stock,
            "market": market,
            "data": df.to_dict('records'),
            "count": len(df)
        }
    except Exception as e:
        return {"error": f"获取个股资金流向失败: {str(e)}"}

def main():
    parser = argparse.ArgumentParser(description='AKShare 特色数据获取')
    parser.add_argument('--type', required=True,
                       choices=['lhb_detail', 'lhb_hyyyb', 'margin_sse', 'margin_detail',
                               'hsgt_hist', 'hsgt_hold', 'gdfx_top10', 'gdfx_free_top10',
                               'board_industry', 'board_concept', 'board_cons', 'restricted_release',
                               'market_fund_flow', 'individual_fund_flow'],
                       help='数据类型')
    parser.add_argument('--start', help='开始日期 YYYYMMDD')
    parser.add_argument('--end', help='结束日期 YYYYMMDD')
    parser.add_argument('--date', help='指定日期 YYYYMMDD')
    parser.add_argument('--symbol', help='股票代码或板块名称')
    parser.add_argument('--market', default='sz', help='市场类型')
    parser.add_argument('--indicator', default='今日排行', help='指标类型')

    args = parser.parse_args()

    if args.type == 'lhb_detail':
        if not args.start or not args.end:
            result = {"error": "龙虎榜详细数据需要start和end参数"}
        else:
            result = get_stock_lhb_detail(args.start, args.end)
    elif args.type == 'lhb_hyyyb':
        if not args.start or not args.end:
            result = {"error": "龙虎榜营业部排名需要start和end参数"}
        else:
            result = get_stock_lhb_hyyyb(args.start, args.end)
    elif args.type == 'margin_sse':
        if not args.start or not args.end:
            result = {"error": "融资融券汇总数据需要start和end参数"}
        else:
            result = get_stock_margin_sse(args.start, args.end)
    elif args.type == 'margin_detail':
        if not args.date:
            result = {"error": "融资融券明细需要date参数"}
        else:
            result = get_stock_margin_detail(args.date)
    elif args.type == 'hsgt_hist':
        symbol = args.symbol if args.symbol else "北向资金"
        result = get_stock_hsgt_hist(symbol)
    elif args.type == 'hsgt_hold':
        market = args.market if args.market else "北向"
        indicator = args.indicator if args.indicator else "今日排行"
        result = get_stock_hsgt_hold_stock(market, indicator)
    elif args.type == 'gdfx_top10':
        if not args.symbol or not args.date:
            result = {"error": "前十大股东需要symbol和date参数"}
        else:
            result = get_stock_gdfx_top_10(args.symbol, args.date)
    elif args.type == 'gdfx_free_top10':
        if not args.symbol or not args.date:
            result = {"error": "前十大流通股东需要symbol和date参数"}
        else:
            result = get_stock_gdfx_free_top_10(args.symbol, args.date)
    elif args.type == 'board_industry':
        result = get_stock_board_industry()
    elif args.type == 'board_concept':
        result = get_stock_board_concept()
    elif args.type == 'board_cons':
        symbol = args.symbol if args.symbol else "银行"
        result = get_stock_board_industry_cons(symbol)
    elif args.type == 'restricted_release':
        symbol = args.symbol if args.symbol else "全部A股"
        result = get_stock_restricted_release(symbol)
    elif args.type == 'market_fund_flow':
        result = get_stock_market_fund_flow()
    elif args.type == 'individual_fund_flow':
        if not args.symbol:
            result = {"error": "个股资金流向需要symbol参数"}
        else:
            market = args.market if args.market else "sz"
            result = get_stock_individual_fund_flow(args.symbol, market)
    else:
        result = {"error": f"未知类型: {args.type}"}

    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))

if __name__ == "__main__":
    main()