Jqdatasdk

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
聚宽数据SDK认证模块
支持用户名密码和Token两种认证方式
支持自定义 host 和 port
"""

import json
import sys
import os


def auth_jq():
    """
    认证并返回已认证的 jqdatasdk 模块
    
    环境变量优先级:
    1. JQDATA_USERNAME + JQDATA_PASSWORD (用户名密码方式)
    2. JQDATA_TOKEN (Token方式)
    
    可选环境变量:
    - JQDATA_HOST: 自定义服务器地址
    - JQDATA_PORT: 自定义端口
    
    Returns:
        module: 已认证的 jqdatasdk 模块
        
    Exits:
        如果认证失败，输出JSON错误信息并退出
    """
    try:
        import jqdatasdk as jq
    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'jqdatasdk 库未安装，请先安装: pip install jqdatasdk'
        }, ensure_ascii=False))
        sys.exit(1)
    
    # 获取可选的 host 和 port 配置
    host = os.getenv('JQDATA_HOST') or None
    port = os.getenv('JQDATA_PORT') or None
    if port:
        try:
            port = int(port)
        except ValueError:
            print(json.dumps({
                'status': 'error',
                'message': f'JQDATA_PORT 必须是有效的端口号: {port}'
            }, ensure_ascii=False))
            sys.exit(1)
    
    # 优先使用用户名密码方式
    username = os.getenv('JQDATA_USERNAME')
    password = os.getenv('JQDATA_PASSWORD')
    
    if username and password:
        try:
            jq.auth(username, password, host=host, port=port)
            return jq
        except Exception as e:
            print(json.dumps({
                'status': 'error',
                'message': f'聚宽认证失败(用户名密码方式): {str(e)}'
            }, ensure_ascii=False))
            sys.exit(1)
    
    # 回退到Token方式，使用正确的 auth_by_token 方法
    token = os.getenv('JQDATA_TOKEN')
    if token:
        try:
            jq.auth_by_token(token, host=host, port=port)
            return jq
        except Exception as e:
            print(json.dumps({
                'status': 'error',
                'message': f'聚宽认证失败(Token方式): {str(e)}'
            }, ensure_ascii=False))
            sys.exit(1)
    
    # 两种认证方式都未配置
    print(json.dumps({
        'status': 'error',
        'message': '缺少聚宽认证凭证，请配置以下任一方式: \n1. JQDATA_USERNAME + JQDATA_PASSWORD (推荐)\n2. JQDATA_TOKEN'
    }, ensure_ascii=False))
    sys.exit(1)


if __name__ == '__main__':
    # 测试认证
    jq = auth_jq()
    print(json.dumps({
        'status': 'success',
        'message': '聚宽认证成功'
    }, ensure_ascii=False))

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
因子数据获取脚本
支持获取估值因子、情绪因子、技术因子、成长因子等
"""

import argparse
import json
import sys
import os
from datetime import datetime, timedelta

# 添加脚本目录到 Python 路径，支持独立运行
script_dir = os.path.dirname(os.path.abspath(__file__))
if script_dir not in sys.path:
    sys.path.insert(0, script_dir)

# 导入认证模块
from auth import auth_jq


def main():
    parser = argparse.ArgumentParser(description='获取聚宽因子数据')
    parser.add_argument('--code', required=True, help='股票代码，格式: 000001.XSHE 或 600000.XSHG，多个用逗号分隔')
    parser.add_argument('--factor', required=True,
                       choices=['valuation', 'turnover', 'basic', 'technical', 'growth', 'momentum', 'size'],
                       help='因子类型: valuation-估值, turnover-换手率, basic-基础, technical-技术, growth-成长, momentum-动量, size-规模')
    parser.add_argument('--start', help='开始日期，格式YYYY-MM-DD，如 2024-01-01')
    parser.add_argument('--end', help='结束日期，格式YYYY-MM-DD，如 2024-12-31')

    args = parser.parse_args()

    try:
        # 使用统一认证模块进行认证
        jq = auth_jq()

        # 设置默认日期范围
        if not args.start:
            args.start = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
        if not args.end:
            args.end = datetime.now().strftime('%Y-%m-%d')

        codes = args.code.split(',')

        result = {
            'status': 'success',
            'factor_type': args.factor,
            'code': args.code,
            'start_date': args.start,
            'end_date': args.end,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }

        if args.factor == 'valuation':
            # 估值因子
            query = jq.query(jq.valuation.code,
                            jq.valuation.day,
                            jq.valuation.market_cap,
                            jq.valuation.circulating_market_cap,
                            jq.valuation.pe_ratio,
                            jq.valuation.pb_ratio,
                            jq.valuation.pcf_ratio,
                            jq.valuation.ps_ratio,
                            jq.valuation.turnover_ratio)

            query = query.filter(jq.valuation.code.in_(codes))
            query = query.filter(jq.valuation.day >= args.start)
            query = query.filter(jq.valuation.day <= args.end)

            df = jq.finance.run_query(query)

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

        elif args.factor == 'turnover':
            # 换手率因子
            query = jq.query(jq.valuation.code,
                            jq.valuation.day,
                            jq.valuation.turnover_ratio,
                            jq.valuation.circulating_market_cap)

            query = query.filter(jq.valuation.code.in_(codes))
            query = query.filter(jq.valuation.day >= args.start)
            query = query.filter(jq.valuation.day <= args.end)

            df = jq.finance.run_query(query)

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

        elif args.factor == 'basic':
            # 基础因子
            query = jq.query(jq.valuation.code,
                            jq.valuation.day,
                            jq.valuation.market_cap,
                            jq.valuation.circulating_market_cap,
                            jq.valuation.pe_ratio,
                            jq.valuation.pb_ratio)

            query = query.filter(jq.valuation.code.in_(codes))
            query = query.filter(jq.valuation.day >= args.start)
            query = query.filter(jq.valuation.day <= args.end)

            df = jq.finance.run_query(query)

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

        elif args.factor == 'technical':
            # 技术因子（通过价格数据计算）
            prices = jq.get_price(
                security=codes,
                start_date=args.start,
                end_date=args.end,
                frequency='daily',
                fields=['open', 'close', 'high', 'low', 'volume', 'money']
            )

            if prices is not None and not prices.empty:
                # 计算简单技术指标
                import pandas as pd
                df = prices.reset_index()

                # 按代码分组计算
                result_list = []
                for code in codes:
                    code_df = df[df['code'] == code].copy()
                    if not code_df.empty:
                        code_df = code_df.sort_values('time')
                        # 计算5日、20日移动平均
                        code_df['ma5'] = code_df['close'].rolling(window=5).mean()
                        code_df['ma20'] = code_df['close'].rolling(window=20).mean()
                        # 计算振幅
                        code_df['amplitude'] = (code_df['high'] - code_df['low']) / code_df['close'] * 100
                        # 计算涨跌幅
                        code_df['change_pct'] = code_df['close'].pct_change() * 100

                        result_list.extend(code_df.to_dict('records'))

                result['data'] = result_list
                result['count'] = len(result_list)
            else:
                result['data'] = []
                result['count'] = 0

        elif args.factor == 'growth':
            # 成长因子（需要财务数据）
            query = jq.query(jq.indicator.code,
                            jq.indicator.statDate,
                            jq.indicator.inc_return,
                            jq.indicator.inc_revenue_year_on_year,
                            jq.indicator.roe)

            query = query.filter(jq.indicator.code.in_(codes))

            if args.start:
                query = query.filter(jq.indicator.statDate >= args.start.replace('-', ''))
            if args.end:
                query = query.filter(jq.indicator.statDate <= args.end.replace('-', ''))

            df = jq.finance.run_query(query)

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

        elif args.factor == 'momentum':
            # 动量因子（通过价格数据计算）
            prices = jq.get_price(
                security=codes,
                start_date=args.start,
                end_date=args.end,
                frequency='daily',
                fields=['close', 'volume']
            )

            if prices is not None and not prices.empty:
                import pandas as pd
                df = prices.reset_index()

                result_list = []
                for code in codes:
                    code_df = df[df['code'] == code].copy()
                    if not code_df.empty:
                        code_df = code_df.sort_values('time')
                        # 计算5日、20日、60日收益率
                        code_df['return_5d'] = code_df['close'].pct_change(5) * 100
                        code_df['return_20d'] = code_df['close'].pct_change(20) * 100
                        code_df['return_60d'] = code_df['close'].pct_change(60) * 100
                        # 计算波动率
                        code_df['volatility'] = code_df['close'].pct_change().rolling(window=20).std() * 100

                        result_list.extend(code_df.to_dict('records'))

                result['data'] = result_list
                result['count'] = len(result_list)
            else:
                result['data'] = []
                result['count'] = 0

        elif args.factor == 'size':
            # 规模因子
            query = jq.query(jq.valuation.code,
                            jq.valuation.day,
                            jq.valuation.market_cap,
                            jq.valuation.circulating_market_cap)

            query = query.filter(jq.valuation.code.in_(codes))
            query = query.filter(jq.valuation.day >= args.start)
            query = query.filter(jq.valuation.day <= args.end)

            df = jq.finance.run_query(query)

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

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'jqdatasdk 库未安装，请先安装: pip install jqdatasdk'
        }, 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

# 添加脚本目录到 Python 路径，支持独立运行
script_dir = os.path.dirname(os.path.abspath(__file__))
if script_dir not in sys.path:
    sys.path.insert(0, script_dir)

# 导入认证模块
from auth import auth_jq


def main():
    parser = argparse.ArgumentParser(description='获取聚宽财务数据')
    parser.add_argument('--type', required=True, choices=['fundamentals', 'indicators', 'valuation'],
                       help='数据类型: fundamentals-财务数据, indicators-指标数据, valuation-估值数据')
    parser.add_argument('--code', required=True, help='股票代码，格式: 000001.XSHE 或 600000.XSHG，多个用逗号分隔')
    parser.add_argument('--statDate', help='报告期，格式YYYY或YYYY-MM-DD，如 2024 或 2024-03-31')
    parser.add_argument('--columns', help='字段列表，逗号分隔，如 pe_ratio, pb_ratio')

    args = parser.parse_args()

    try:
        # 使用统一认证模块进行认证
        jq = auth_jq()

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

        codes = args.code.split(',')

        if args.type == 'fundamentals':
            # 财务数据
            query = jq.query(jq.finance.STK_BALANCE_SHEET)

            if args.statDate:
                query = query.filter(jq.finance.STK_BALANCE_SHEET.statDate == args.statDate)

            query = query.filter(jq.finance.STK_BALANCE_SHEET.code.in_(codes))

            if args.columns:
                columns = args.columns.split(',')
                query = query.filter(*[getattr(jq.finance.STK_BALANCE_SHEET, col) != None for col in columns])

            df = jq.finance.run_query(query)

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

        elif args.type == 'indicators':
            # 指标数据
            query = jq.query(jq.indicator)

            if args.statDate:
                query = query.filter(jq.indicator.statDate == args.statDate)

            query = query.filter(jq.indicator.code.in_(codes))

            df = jq.finance.run_query(query)

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

        elif args.type == 'valuation':
            # 估值数据
            query = jq.query(jq.valuation)

            if args.statDate:
                query = query.filter(jq.valuation.day == args.statDate)
            else:
                # 获取最新估值数据
                query = query.order_by(jq.valuation.day.desc())

            query = query.filter(jq.valuation.code.in_(codes))

            if args.columns:
                selected_columns = ['code', 'day'] + args.columns.split(',')
                df = jq.finance.run_query(query)
                if df is not None and not df.empty:
                    df = df[selected_columns]
            else:
                df = jq.finance.run_query(query)

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

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'jqdatasdk 库未安装，请先安装: pip install jqdatasdk'
        }, 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 -*-
"""
行情数据获取脚本
支持获取价格、K线、实时数据、Tick数据
"""

import argparse
import json
import sys
import os
from datetime import datetime, timedelta

# 添加脚本目录到 Python 路径，支持独立运行
script_dir = os.path.dirname(os.path.abspath(__file__))
if script_dir not in sys.path:
    sys.path.insert(0, script_dir)

# 导入认证模块
from auth import auth_jq


def main():
    parser = argparse.ArgumentParser(description='获取聚宽A股行情数据')
    parser.add_argument('--type', required=True, choices=['price', 'current', 'ticks', 'order', 'future_ticks'],
                       help='数据类型: price-价格/K线, current-实时数据, ticks-Tick, order-委托, future_ticks-期货Tick')
    parser.add_argument('--code', required=True, help='股票代码，格式: 000001.XSHE 或 600000.XSHG')
    parser.add_argument('--start', help='开始日期，格式YYYY-MM-DD，如 2024-01-01')
    parser.add_argument('--end', help='结束日期，格式YYYY-MM-DD，如 2024-01-31')
    parser.add_argument('--frequency', default='daily',
                       choices=['1min', '5min', '15min', '30min', '60min', '120min', 'daily', 'weekly', 'monthly'],
                       help='数据频率，默认daily')
    parser.add_argument('--count', type=int, default=100, help='获取数量，默认100')
    parser.add_argument('--fields', help='字段列表，逗号分隔，如 open,close,high,low,volume')

    args = parser.parse_args()

    try:
        # 使用统一认证模块进行认证
        jq = auth_jq()

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

        if args.type == 'price':
            if not args.start:
                args.end = datetime.now().strftime('%Y-%m-%d')
                args.start = (datetime.now() - timedelta(days=args.count)).strftime('%Y-%m-%d')

            # 默认字段
            default_fields = ['open', 'close', 'high', 'low', 'volume', 'money']
            fields = args.fields.split(',') if args.fields else default_fields

            data = jq.get_price(
                security=args.code,
                start_date=args.start,
                end_date=args.end,
                frequency=args.frequency,
                fields=fields
            )

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

        elif args.type == 'current':
            if ',' in args.code:
                codes = args.code.split(',')
                data = jq.get_current_data()
                result['data'] = []
                if isinstance(data, dict):
                    for code in codes:
                        if code in data:
                            result['data'].append({
                                'code': code,
                                'last_price': data[code].last_price,
                                'time': data[code].time.strftime('%Y-%m-%d %H:%M:%S') if hasattr(data[code].time, 'strftime') else str(data[code].time),
                                'day_open': data[code].day_open,
                                'day_high': data[code].day_high,
                                'day_low': data[code].day_low,
                                'volume': data[code].volume,
                                'money': data[code].money
                            })
                result['count'] = len(result['data'])
            else:
                data = jq.get_current_data()
                if isinstance(data, dict) and args.code in data:
                    result['data'] = {
                        'code': args.code,
                        'last_price': data[args.code].last_price,
                        'time': data[args.code].time.strftime('%Y-%m-%d %H:%M:%S') if hasattr(data[args.code].time, 'strftime') else str(data[args.code].time),
                        'day_open': data[args.code].day_open,
                        'day_high': data[args.code].day_high,
                        'day_low': data[args.code].day_low,
                        'volume': data[args.code].volume,
                        'money': data[args.code].money
                    }
                else:
                    result['error'] = f'未获取到股票 {args.code} 的实时数据'

        elif args.type == 'ticks':
            if not args.start:
                args.end = datetime.now().strftime('%Y-%m-%d')
                args.start = args.end

            data = jq.get_ticks(
                security=args.code,
                start_dt=args.start,
                end_dt=args.end
            )

            tick_data = []
            if data:
                for tick in data:
                    tick_data.append({
                        'time': tick.time.strftime('%Y-%m-%d %H:%M:%S') if hasattr(tick.time, 'strftime') else str(tick.time),
                        'price': tick.price,
                        'volume': tick.volume,
                        'amount': tick.amount,
                        'direction': tick.direction if hasattr(tick, 'direction') else ''
                    })
            result['ticks'] = tick_data
            result['count'] = len(tick_data)

        elif args.type == 'order':
            if not args.start:
                args.end = datetime.now().strftime('%Y-%m-%d')
                args.start = args.end

            data = jq.get_order_book(
                security=args.code,
                start_dt=args.start,
                end_dt=args.end
            )

            order_data = []
            if data:
                for order in data:
                    order_data.append({
                        'time': order.time.strftime('%Y-%m-%d %H:%M:%S') if hasattr(order.time, 'strftime') else str(order.time),
                        'b_p_1': getattr(order, 'b_p_1', 0),
                        'b_v_1': getattr(order, 'b_v_1', 0),
                        'a_p_1': getattr(order, 'a_p_1', 0),
                        'a_v_1': getattr(order, 'a_v_1', 0),
                        'b_p_2': getattr(order, 'b_p_2', 0),
                        'b_v_2': getattr(order, 'b_v_2', 0),
                        'a_p_2': getattr(order, 'a_p_2', 0),
                        'a_v_2': getattr(order, 'a_v_2', 0),
                        'b_p_3': getattr(order, 'b_p_3', 0),
                        'b_v_3': getattr(order, 'b_v_3', 0),
                        'a_p_3': getattr(order, 'a_p_3', 0),
                        'a_v_3': getattr(order, 'a_v_3', 0)
                    })
            result['orders'] = order_data
            result['count'] = len(order_data)

        elif args.type == 'future_ticks':
            if not args.start:
                args.end = datetime.now().strftime('%Y-%m-%d')
                args.start = args.end

            data = jq.get_future_ticks(
                security=args.code,
                start_dt=args.start,
                end_dt=args.end
            )

            tick_data = []
            if data:
                for tick in data:
                    tick_data.append({
                        'time': tick.time.strftime('%Y-%m-%d %H:%M:%S') if hasattr(tick.time, 'strftime') else str(tick.time),
                        'price': tick.price,
                        'volume': tick.volume,
                        'amount': tick.amount
                    })
            result['ticks'] = tick_data
            result['count'] = len(tick_data)

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'jqdatasdk 库未安装，请先安装: pip install jqdatasdk'
        }, 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

# 添加脚本目录到 Python 路径，支持独立运行
script_dir = os.path.dirname(os.path.abspath(__file__))
if script_dir not in sys.path:
    sys.path.insert(0, script_dir)

# 导入认证模块
from auth import auth_jq


def main():
    parser = argparse.ArgumentParser(description='获取聚宽股票信息')
    parser.add_argument('--type', required=True, choices=['all_securities', 'get_security_info', 'trade_days'],
                       help='类型: all_securities-获取证券列表, get_security_info-获取证券信息, trade_days-获取交易日')
    parser.add_argument('--security_type', default='stock',
                       choices=['stock', 'fund', 'index', 'etf', 'futures', 'all'],
                       help='证券类型，默认stock')
    parser.add_argument('--code', help='证券代码，格式: 000001.XSHE 或 600000.XSHG')
    parser.add_argument('--start', help='开始日期，格式YYYY-MM-DD，如 2024-01-01')
    parser.add_argument('--end', help='结束日期，格式YYYY-MM-DD，如 2024-12-31')

    args = parser.parse_args()

    try:
        # 使用统一认证模块进行认证
        jq = auth_jq()

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

        if args.type == 'all_securities':
            # 获取证券列表
            if args.security_type == 'all':
                securities = jq.get_all_securities()
            else:
                securities = jq.get_all_securities(types=[args.security_type])

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

        elif args.type == 'get_security_info':
            if not args.code:
                result['error'] = '请提供 --code 参数'
            else:
                info = jq.get_security_info(args.code)
                if info:
                    result['data'] = {
                        'code': info.code,
                        'display_name': info.display_name,
                        'name': info.name,
                        'start_date': info.start_date.strftime('%Y-%m-%d') if hasattr(info.start_date, 'strftime') else str(info.start_date),
                        'end_date': info.end_date.strftime('%Y-%m-%d') if info.end_date and hasattr(info.end_date, 'strftime') else str(info.end_date) if info.end_date else None,
                        'type': info.type,
                        'parent': info.parent
                    }
                else:
                    result['error'] = f'未找到证券 {args.code} 的信息'

        elif args.type == 'trade_days':
            if not args.start or not args.end:
                result['error'] = '请提供 --start 和 --end 参数'
            else:
                trade_days = jq.get_trade_days(start_date=args.start, end_date=args.end)
                result['trade_days'] = [day.strftime('%Y-%m-%d') for day in trade_days] if trade_days else []
                result['count'] = len(result['trade_days']) if result['trade_days'] else 0

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

    except ImportError:
        print(json.dumps({
            'status': 'error',
            'message': 'jqdatasdk 库未安装，请先安装: pip install jqdatasdk'
        }, 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()