eta_python/ths_api.py

import hug
from bottle import route, run, NORUN
import iFinDPy as THS
import json
import time
import logging

logger = logging.getLogger('ths_api')
logger.setLevel(logging.DEBUG)  # 设置日志级别

hug.API(__name__).http.output_format = hug.output_format.json

username = ""
password = ""

file_handler = logging.FileHandler('ths_api.log')
file_handler.setLevel(logging.DEBUG)

formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)

logger.addHandler(file_handler)


@hug.get('/hz_server')
def hello():
    return 1
    # return 'wind true'


# 同花顺登录函数
def thslogin():
    # 输入用户的帐号和密码
    thsLogin = THS.THS_iFinDLogin(username, password)
    print(thsLogin)
    if (thsLogin == 0 or thsLogin == -201):
        print('ths 登录成功')
        logger.info('ths login success')
    else:
        print('ths 登录失败')
        logger.info('ths login fail')


@hug.get('/edbInfo/ths')
def GetEdbDataByThs(EdbCode, StartDate, EndDate):
    print("GetEdbDataByThs")
    logger.info('GetEdbDataByThs')
    thsEDBDataQuery = THS.THS_EDBQuery(EdbCode, StartDate, EndDate, True)
    print(thsEDBDataQuery)
    logger.info(thsEDBDataQuery)

    print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ":start GetEdbDataByThs")
    print("thsEDBDataQuery")
    print(thsEDBDataQuery)
    thsEDBDataQuery = thsEDBDataQuery.decode("unicode_escape")

    result = json.loads(thsEDBDataQuery)
    print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ":end GetEdbDataByThs")

    if result["errorcode"] == -1010:
        thsLogin = THS.THS_iFinDLogin(username, password)
        print("登录一次")
        print(thsLogin)
        logger.info("ths login")
        if (thsLogin == 0 or thsLogin == -201):
            thsEDBDataQuery = THS.THS_EDBQuery(EdbCode, StartDate, EndDate, True)
            thsEDBDataQuery = thsEDBDataQuery.decode("unicode_escape")
            result = json.loads(thsEDBDataQuery)
            logger.info("login success")
    else:
        print("已登录")

    return result


@hug.get('/edbInfo/ths/future_good')
def GetFutureGoodEdbDataByThs(EdbCode, StartDate, EndDate):
    print("THS_HistoryQuotes start")
    logger.info('THS_HistoryQuotes get')
    filed = 'lastclose,open,high,low,close,avgprice,change,changeper,volume,amount,hsl,lastsettlement,settlement,zdsettlement,zdfsettlement,ccl,ccbd,zf,zjlx,zjcd'
    # filed = 'open,high,low,close,volume,amount,settlement,ccl,zjlx'
    params = 'Interval:D,CPS:1,baseDate:1900-01-01,Currency:YSHB,fill:Previous'
    # params = 'YSHB;Tradedays'
    # print(params)
    thsEDBDataQuery = THS.THS_HistoryQuotes(EdbCode, filed, params, StartDate, EndDate, True)
    # thsEDBDataQuery = THS.THS_HQ(EdbCode, filed, params, StartDate, EndDate)
    # thsEDBDataQuery = THS.THS_EDBQuery(EdbCode, StartDate, EndDate, True)
    print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ":start GetFutureGoodEdbDataByThs")
    print("THS_HistoryQuotes end")
    print(thsEDBDataQuery)
    logger.info(thsEDBDataQuery)
    thsEDBDataQuery = thsEDBDataQuery.decode("unicode_escape")
    result = json.loads(thsEDBDataQuery)
    if result["errorcode"] == -1010:
        thsLogin = THS.THS_iFinDLogin(username, password)
        print("登录一次")
        print(thsLogin)
        logger.info('THS_HistoryQuotes login')
        if (thsLogin == 0 or thsLogin == -201):
            thsEDBDataQuery = THS.THS_HistoryQuotes(EdbCode, filed, params, StartDate, EndDate, True)
            thsEDBDataQuery = thsEDBDataQuery.decode("unicode_escape")
            result = json.loads(thsEDBDataQuery)
    else:
        print("已登录")
    print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ":end GetFutureGoodEdbDataByThs")
    return result


def ths_data_to_dict(ths_instance):
    if ths_instance is not None and ths_instance.data is not None:
        return {
            'errorcode': ths_instance.errorcode,
            'errmsg': ths_instance.errmsg,
            'data': ths_instance.data.to_dict(orient='records')
        }
    else:
        # 处理 ths_instance 为 None 的情况
        return {
            'errorcode': 'N/A',
            'errmsg': 'N/A',
            'data': []
        }


def has_comma(input_string):
    return isinstance(input_string, str) and ',' in input_string


def process_single_edb_code(StockCode, SingleEdbCode, StartDate, EndDate, Days='', Interval='', ExtraPars=''):
    strExtra = ','.join(map(str, ExtraPars))
    print(strExtra)

    funcParas = []
    if Days:
        funcParas.append(f"Days:{Days}")
    if Interval:
        funcParas.append(f"Interval:{Interval}")
    funcPara = ",".join(funcParas)

    thsEDBDataQuery = THS.THS_DS(StockCode, SingleEdbCode, strExtra, funcPara, StartDate, EndDate)
    print(thsEDBDataQuery)
    result = json.dumps(ths_data_to_dict(thsEDBDataQuery))
    return result


@hug.get('/edbInfo/ths/ds')
def GetEdbDataByThsDs(StockCode, EdbCode, StartDate, EndDate, Days='', Interval='', ExtraPars=''):
    print(
        "GetEdbDataByThsDs-> StockCode: {}; EdbCode: {}; StartDate: {}; EndDate: {}; Days: {}; Interval: {}; ExtraPars: {}".format(
            StockCode,
            EdbCode,
            StartDate,
            EndDate,
            Days,
            Interval,
            ExtraPars))
    result_list = []

    if isinstance(EdbCode, list):
        for SingleEdbCode in EdbCode:
            result_list.append(
                process_single_edb_code(StockCode, SingleEdbCode, StartDate, EndDate, Days, Interval, ExtraPars))
            print("THS_DS end")
        print(result_list)
        return result_list
    else:
        # 如果不包含逗号，直接处理
        result = process_single_edb_code(StockCode, EdbCode, StartDate, EndDate, Days, Interval, ExtraPars)
        print(result)
        return result


@hug.get('/edbInfo/ths/hf')
def GetEdbDataByThsHf(codes, indicators, function_para, start_time, end_time):
    print("codes: {}; indicators: {}; function_para: {}; start_time: {}; end_time: {}".format(codes, indicators,
                                                                                              function_para, start_time,
                                                                                              end_time))
    data_query = THS.THS_HF(codes, indicators, function_para, start_time, end_time)
    print("data_query: {}".format(data_query))

    if data_query.errorcode == -1010 or data_query.errorcode == -209:
        ths_login = THS.THS_iFinDLogin(username, password)
        print("登录一次")
        print("ths_login: {}".format(ths_login))

        if ths_login == 0 or ths_login == -201:
            repeat_query = THS.THS_HF(codes, indicators, function_para, start_time, end_time)
            print("repeat_query: {}".format(repeat_query))

            return json.dumps(ths_data_to_dict(repeat_query), indent=2)
    else:
        return json.dumps(ths_data_to_dict(data_query), indent=2)


if __name__ == "__main__":
    # ths登录函数
    thslogin()
    app = __hug__.http.server()
    run(app=app, reloader=True, host='0.0.0.0', port=7000)