Tushare

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
财务数据获取脚本
支持获取财务数据、利润表、资产负债表、现金流量表、财务指标
"""

import argparse
import json
import sys
import os
from datetime import datetime


def main():
    parser = argparse.ArgumentParser(description='获取 Tushare 财务数据')
    parser.add_argument('--type', required=True,
                       choices=['financial', 'income', 'balancesheet', 'cashflow', 'fina_indicator', 'fina_audit', 'fina_mainbz'],
                       help='数据类型: financial-财务数据, income-利润表, balancesheet-资产负债表, cashflow-现金流量表, fina_indicator-财务指标, fina_audit-审计意见, fina_mainbz-主营业务')
    parser.add_argument('--ts_code', help='股票代码，格式: 000001.SZ 或 600000.SH，多个用逗号分隔')
    parser.add_argument('--period', help='报告期，格式YYYY或YYYYMM，如 2023 或 202312')
    parser.add_argument('--start_date', help='开始日期，格式YYYYMMDD')
    parser.add_argument('--end_date', help='结束日期，格式YYYYMMDD')
    parser.add_argument('--report_type', help='报告类型: 1-合并报表, 2-单季合并, 3-调整单季合并表, 4-调整合并报表, 5-调整前合并报表')
    parser.add_argument('--fields', help='字段列表，逗号分隔')

    args = parser.parse_args()

    try:
        # 获取凭证
        credential = os.getenv(f"TUSHARE_TOKEN")
        if not credential:
            print(json.dumps({
                'status': 'error',
                'message': '缺少 Tushare Token 凭证配置，请先配置 TUSHARE_TOKEN 凭证'
            }, ensure_ascii=False))
            sys.exit(1)

        import tushare as ts

        # 初始化
        pro = ts.pro_api(credential)

        result = {
            'status': 'success',
            'data_type': args.type,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }

        params = {}
        if args.ts_code:
            params['ts_code'] = args.ts_code
        if args.period:
            params['period'] = args.period
        if args.start_date:
            params['start_date'] = args.start_date
        if args.end_date:
            params['end_date'] = args.end_date
        if args.report_type:
            params['report_type'] = args.report_type
        if args.fields:
            params['fields'] = args.fields

        if args.type == 'financial':
            # 财务数据（旧接口）
            data = pro.financial(**params)

        elif args.type == 'income':
            # 利润表
            data = pro.income(**params)

        elif args.type == 'balancesheet':
            # 资产负债表
            data = pro.balancesheet(**params)

        elif args.type == 'cashflow':
            # 现金流量表
            data = pro.cashflow(**params)

        elif args.type == 'fina_indicator':
            # 财务指标
            data = pro.fina_indicator(**params)

        elif args.type == 'fina_audit':
            # 审计意见
            data = pro.fina_audit(**params)

        elif args.type == 'fina_mainbz':
            # 主营业务
            data = pro.fina_mainbz(**params)

        if data is not None and not data.empty:
            result['data'] = data.to_dict('records')
            result['count'] = len(data)
        else:
            result['data'] = []
            result['count'] = 0

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'tushare 库未安装，请先安装: pip install tushare'
        }, ensure_ascii=False))
        sys.exit(1)
    except Exception as e:
        print(json.dumps({
            'status': 'error',
            'message': str(e),
            'error_type': type(e).__name__
        }, ensure_ascii=False))
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
宏观数据与指数数据获取脚本
支持获取指数行情、指数基础信息、Shibor 利率等
"""

import argparse
import json
import sys
import os
from datetime import datetime


def main():
    parser = argparse.ArgumentParser(description='获取 Tushare 指数与宏观数据')
    parser.add_argument('--type', required=True,
                       choices=['index_daily', 'index_basic', 'index_classify', 'index_daily_basic', 'shibor', 'shibor_rate', 'shibor_quote'],
                       help='类型: index_daily-指数行情, index_basic-指数基础信息, index_classify-指数分类, index_daily_basic-指数每日指标, shibor-Shibor, shibor_rate-Shibor利率, shibor_quote-Shibor报价')
    parser.add_argument('--ts_code', help='指数代码，格式: 000001.SH 或 399001.SZ')
    parser.add_argument('--start_date', help='开始日期，格式YYYYMMDD')
    parser.add_argument('--end_date', help='结束日期，格式YYYYMMDD')
    parser.add_argument('--trade_date', help='交易日，格式YYYYMMDD')
    parser.add_argument('--market', help='市场:主板/中小板/创业板/科创板等')
    def main():
    parser.add_argument('--level', help='级别: L1-大, L2-中, L3-小')

    args = parser.parse_args()

    try:
        # 获取凭证
        credential = os.getenv(f"TUSHARE_TOKEN")
        if not credential:
            print(json.dumps({
                'status': 'error',
                'message': '缺少 Tushare Token 凭证配置，请先配置 TUSHARE_TOKEN 凭证'
            }, ensure_ascii=False))
            sys.exit(1)

        import tushare as ts

        # 初始化
        pro = ts.pro_api(credential)

        result = {
            'status': 'success',
            'data_type': args.type,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }

        params = {}

        if args.type == 'index_daily':
            # 指数行情
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date

            data = pro.index_daily(**params)

        elif args.type == 'index_basic':
            # 指数基础信息
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.market:
                params['market'] = args.market
            if args.level:
                params['level'] = args.level

            data = pro.index_basic(**params)

        elif args.type == 'index_classify':
            # 指数分类
            if args.market:
                params['market'] = args.market
            if args.level:
                params['level'] = args.level

            data = pro.index_classify(**params)

        elif args.type == 'index_daily_basic':
            # 指数每日指标
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date

            data = pro.index_daily_basic(**params)

        elif args.type == 'shibor':
            # Shibor 数据
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date

            data = pro.shibor(**params)

        elif args.type == 'shibor_rate':
            # Shibor 利率
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date

            data = pro.shibor_rate(**params)

        elif args.type == 'shibor_quote':
            # Shibor 报价
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date

            data = pro.shibor_quote(**params)

        if data is not None and not data.empty:
            result['data'] = data.to_dict('records')
            result['count'] = len(data)
        else:
            result['data'] = []
            result['count'] = 0

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'tushare 库未安装，请先安装: pip install tushare'
        }, ensure_ascii=False))
        sys.exit(1)
    except Exception as e:
        print(json.dumps({
            'status': 'error',
            'message': str(e),
            'error_type': type(e).__name__
        }, ensure_ascii=False))
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
行情数据获取脚本
支持获取日线行情、复权因子、分钟线数据
"""

import argparse
import json
import sys
import os
from datetime import datetime


def main():
    parser = argparse.ArgumentParser(description='获取 Tushare 行情数据')
    parser.add_argument('--type', required=True, choices=['daily', 'adj_factor', 'daily_basic', 'weekly', 'monthly'],
                       help='数据类型: daily-日线, adj_factor-复权因子, daily_basic-每日指标, weekly-周线, monthly-月线')
    parser.add_argument('--ts_code', help='股票代码，格式: 000001.SZ 或 600000.SH，多个用逗号分隔')
    parser.add_argument('--start_date', help='开始日期，格式YYYYMMDD，如 20240101')
    parser.add_argument('--end_date', help='结束日期，格式YYYYMMDD，如 20240131')
    parser.add_argument('--trade_date', help='交易日，格式YYYYMMDD，单日数据查询')
    parser.add_argument('--fields', help='字段列表，逗号分隔，如 ts_code,trade_date,open,high,low,close,vol')

    args = parser.parse_args()

    try:
        # 获取凭证
        credential = os.getenv(f"TUSHARE_TOKEN")
        if not credential:
            print(json.dumps({
                'status': 'error',
                'message': '缺少 Tushare Token 凭证配置，请先配置 TUSHARE_TOKEN 凭证'
            }, ensure_ascii=False))
            sys.exit(1)

        import tushare as ts

        # 初始化
        pro = ts.pro_api(credential)

        result = {
            'status': 'success',
            'data_type': args.type,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }

        if args.type == 'daily':
            # 日线行情
            params = {}
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date
            if args.fields:
                params['fields'] = args.fields

            data = pro.daily(**params)

            if data is not None and not data.empty:
                result['data'] = data.to_dict('records')
                result['count'] = len(data)
            else:
                result['data'] = []
                result['count'] = 0

        elif args.type == 'adj_factor':
            # 复权因子
            params = {}
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date

            data = pro.adj_factor(**params)

            if data is not None and not data.empty:
                result['data'] = data.to_dict('records')
                result['count'] = len(data)
            else:
                result['data'] = []
                result['count'] = 0

        elif args.type == 'daily_basic':
            # 每日指标
            params = {}
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date
            if args.fields:
                params['fields'] = args.fields

            data = pro.daily_basic(**params)

            if data is not None and not data.empty:
                result['data'] = data.to_dict('records')
                result['count'] = len(data)
            else:
                result['data'] = []
                result['count'] = 0

        elif args.type == 'weekly':
            # 周线行情
            params = {}
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date
            if args.fields:
                params['fields'] = args.fields

            data = pro.weekly(**params)

            if data is not None and not data.empty:
                result['data'] = data.to_dict('records')
                result['count'] = len(data)
            else:
                result['data'] = []
                result['count'] = 0

        elif args.type == 'monthly':
            # 月线行情
            params = {}
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.trade_date:
                params['trade_date'] = args.trade_date
            if args.fields:
                params['fields'] = args.fields

            data = pro.monthly(**params)

            if data is not None and not data.empty:
                result['data'] = data.to_dict('records')
                result['count'] = len(data)
            else:
                result['data'] = []
                result['count'] = 0

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'tushare 库未安装，请先安装: pip install tushare'
        }, ensure_ascii=False))
        sys.exit(1)
    except Exception as e:
        print(json.dumps({
            'status': 'error',
            'message': str(e),
            'error_type': type(e).__name__
        }, ensure_ascii=False))
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
股票信息获取脚本
支持获取股票列表、交易日历、名称变更等
"""

import argparse
import json
import sys
import os
from datetime import datetime


def main():
    parser = argparse.ArgumentParser(description='获取 Tushare 股票信息')
    parser.add_argument('--type', required=True, choices=['stock_basic', 'trade_cal', 'namechange', 'trade_cal_ext', 'suspend'],
                       help='类型: stock_basic-股票列表, trade_cal-交易日历, namechange-名称变更, trade_cal_ext-交易日历扩展, suspend-停牌信息')
    parser.add_argument('--ts_code', help='股票代码，格式: 000001.SZ 或 600000.SH')
    parser.add_argument('--list_status', help='股票状态: L-上市, D-退市, P-暂停上市')
    parser.add_argument('--exchange', help='交易所: SSE-上交所, SZSE-深交所')
    parser.add_argument('--start_date', help='开始日期，格式YYYYMMDD')
    parser.add_argument('--end_date', help='结束日期，格式YYYYMMDD')
    parser.add_argument('--exchange_id', help='交易所ID: SSE-上交所, SZSE-深交所')
    parser.add_argument('--is_open', help='是否开市: 0-休市, 1-开市')

    args = parser.parse_args()

    try:
        # 获取凭证
        credential = os.getenv(f"TUSHARE_TOKEN")
        if not credential:
            print(json.dumps({
                'status': 'error',
                'message': '缺少 Tushare Token 凭证配置，请先配置 TUSHARE_TOKEN 凭证'
            }, ensure_ascii=False))
            sys.exit(1)

        import tushare as ts

        # 初始化
        pro = ts.pro_api(credential)

        result = {
            'status': 'success',
            'data_type': args.type,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }

        params = {}

        if args.type == 'stock_basic':
            # 股票列表
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.list_status:
                params['list_status'] = args.list_status
            if args.exchange:
                params['exchange'] = args.exchange

            data = pro.stock_basic(**params)

        elif args.type == 'trade_cal':
            # 交易日历
            if args.exchange_id:
                params['exchange'] = args.exchange_id
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.is_open:
                params['is_open'] = args.is_open

            data = pro.trade_cal(**params)

        elif args.type == 'namechange':
            # 名称变更
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date

            data = pro.namechange(**params)

        elif args.type == 'trade_cal_ext':
            # 交易日历扩展
            if args.exchange_id:
                params['exchange'] = args.exchange_id
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date
            if args.is_open:
                params['is_open'] = args.is_open

            data = pro.trade_cal_ext(**params)

        elif args.type == 'suspend':
            # 停牌信息
            if args.ts_code:
                params['ts_code'] = args.ts_code
            if args.start_date:
                params['start_date'] = args.start_date
            if args.end_date:
                params['end_date'] = args.end_date

            data = pro.suspend(**params)

        if data is not None and not data.empty:
            result['data'] = data.to_dict('records')
            result['count'] = len(data)
        else:
            result['data'] = []
            result['count'] = 0

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'tushare 库未安装，请先安装: pip install tushare'
        }, ensure_ascii=False))
        sys.exit(1)
    except Exception as e:
        print(json.dumps({
            'status': 'error',
            'message': str(e),
            'error_type': type(e).__name__
        }, ensure_ascii=False))
        sys.exit(1)


if __name__ == '__main__':
    main()