首页 > 其他分享 >量化交易之One Piece篇 - 基于天勤的数据备用方案(stable版)

量化交易之One Piece篇 - 基于天勤的数据备用方案(stable版)

时间:2023-04-29 11:34:09浏览次数:38  
标签:main datetime self tq future stable Piece data 天勤


from tqsdk import TqApi, TqAuth

from tqsdk.tafunc import time_to_datetime
import datetime
from datetime import datetime

import pandas
import re
import os

import warnings
warnings.filterwarnings('ignore')

pandas.set_option('display.max_columns', None)
pandas.set_option('display.max_rows', None)

from tqz_extern.json_operator import TQZJsonOperator

TIME_GAP = 8 * 60 * 60 * 1000000000


class TQZTianQinClient:
    """
    天勤接口 每次只能拉取单一合约的数据!
    """

    __tq_futures = None

    def __init__(self, account: str = "xxxxxxx", pass_word: str = "xxxxxxx"):
        self.api = TqApi(auth=TqAuth(account, pass_word))

        if TQZTianQinClient.__tq_futures is None:
            TQZTianQinClient.__tq_futures = self.api.query_quotes(ins_class="FUTURE", expired=False)


    def query_history_ticks(self, tq_future: str, tq_data_length: int = 8964):
        assert tq_future in TQZTianQinClient.__tq_futures, f'bad tq_future: {tq_future}'

        tq_result = self.api.get_tick_serial(symbol=tq_future, data_length=tq_data_length)
        self.api.close()

        tq_result['datetime_format'] = tq_result['datetime'].apply(time_to_datetime)

        return tq_result


    def query_single_quote(self, tq_future: str) -> dict:
        result = self.api.get_quote(symbol=tq_future)
        self.api.close()

        return result  # noqa


    def query_history_bars(self, tq_future: str, tq_duration_seconds: int, tq_data_length: int = 8964):
        assert tq_future in TQZTianQinClient.__tq_futures, f'bad tq_future: {tq_future}'

        tq_result = self.api.get_kline_serial(symbol=tq_future, duration_seconds=tq_duration_seconds, data_length=tq_data_length)
        self.api.close()

        tq_result["datetime"] = pandas.to_datetime(tq_result["datetime"] + TIME_GAP)
        tq_result['datetime'] = tq_result['datetime'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))  # %f是毫秒

        return tq_result


    def load_all_tq_futures(self):
        self.api.close()
        return TQZTianQinClient.__tq_futures


    def load_all_tq_main_futures(self):
        tq_main_contracts = self.api.query_quotes(ins_class="CONT")

        main_vt_symbols = []
        [main_vt_symbols.append(
            self.api.get_quote(
                symbol=main_contract
            ).underlying_symbol
        ) for main_contract in tq_main_contracts]

        self.api.close()
        return main_vt_symbols

    @staticmethod
    def record_data():
        api = TqApi(auth=TqAuth('xxxxxxx', 'xxxxxxx'))

        instruments = ["SHFE.cu2305", "DCE.i2309"]

        quote_map = {}
        data_map = {}  # 之后考虑写共享内存里
        for instrument in instruments:
            quote_map[instrument] = api.get_quote(symbol=instrument)
            data_map[instrument] = pandas.DataFrame()

        while api.wait_update():
            for instrument in instruments:
                if api.is_changing(quote_map[instrument]):
                    # print(quote_map[instrument])
                    data_map[instrument].append(vars(quote_map[instrument]), ignore_index=True)

            if datetime.datetime.now().hour == 15:
                today = str(datetime.date.today()).replace("-", "")

                for instrument in instruments:
                    csv_path = f'./config/{instrument.replace(".", "_")}_{today}.csv'

                    if os.path.exists(path=csv_path) is False:
                        data_map[instrument].to_csv(
                            csv_path,
                            index=False
                        )

                break  # 当前脚本, 每天晚盘前需要手工启动脚本


    def __multi_symbols_day_bar(self):
        klines = self.api.get_kline_serial(["SHFE.rb2305", "SHFE.rb2310"], duration_seconds=60 * 60 * 24, data_length=5)
        self.api.close()

        klines_filter = klines.loc[klines['id'] >= 0]

        klines_filter['datetime_format'] = klines_filter['datetime'].apply(time_to_datetime)
        print("klines_filter: " + str(klines_filter))



class TQZAutoMakeConfig:

    __trading_time_config = './config/trading_time.json'

    @classmethod
    def auto_make_session_json(cls):
        all_tq_main_futures = TQZTianQinClient().load_all_tq_main_futures()

        ret = {}
        for tq_main_future in all_tq_main_futures:
            _tq_future = f'{tq_main_future.split(".")[0]}.{re.match(r"^[a-zA-Z]{1,3}", tq_main_future.split(".")[1]).group()}'
            tq_main_future_data = TQZTianQinClient().query_single_quote(tq_future=tq_main_future)

            ret[_tq_future] = {
                'day': vars(tq_main_future_data['trading_time'])['day'],
                'night': vars(tq_main_future_data['trading_time'])['night']
            }

        for instrument, data in ret.items():
            for session_data in data.values():
                for i in range(len(session_data)):
                    for j in range(len(session_data[i])):
                        hour = session_data[i][j].split(':')[0]
                        if hour >= '24':
                            new_hour = str(int(hour) - int('24')).zfill(2)
                            session_data[i][j] = f'{new_hour}:{session_data[i][j].split(":")[1]}:{session_data[i][j].split(":")[2]}'

        TQZJsonOperator.tqz_write_jsonfile(content=ret, target_jsonfile=cls.__trading_time_config)


    @classmethod
    def auto_make_main_contracts_excel(cls):
        """
        根据所有合约的日线收盘数据, 更新当日的所有品种的主力合约
        """

        """ eg: 
                   date |   SHFE.rb   |   SHFE.ag   | ...
               20230426 | SHFE.rb2310 | SHFE.ag2306 | ...
               20230427 | SHFE.rb2310 | SHFE.ag2306 | ...
               20230428 | SHFE.rb2401 | SHFE.ag2312 | ...
        """

        pass



if __name__ == '__main__':
    # ret_df = TQZTianQinClient().query_history_bars(tq_future="SHFE.cu2305", tq_duration_seconds=60)
    # print("ret_df: " + str(ret_df))

    # ret_df = TQZTianQinClient().query_history_ticks(tq_future="SHFE.rb2310")
    # print("ret_df: " + str(ret_df))

    # TQZTianQinClient.record_data()

    # TQZAutoMakeConfig.auto_make_main_contracts_excel()

    pass

标签:main,datetime,self,tq,future,stable,Piece,data,天勤
From: https://blog.51cto.com/u_9527606/6236678

相关文章

  • CF1621A Stable Arrangement of Rooks
    题目简述:一个n*n的棋盘上,放上k个车,使得一任意车向上下左右移动一格(这里的车可以上下左右移动任意步数)后不与其他车相撞(注:不能走出棋盘之外)。个人分析:从题目可知,在车上下左右移动一格后不会与其他车相撞,换句话说,两辆车之间至少相隔一行一列,放在对角线上是最优想法,若无解则......
  • mac使用Stable Diffussion进阶篇
    1、出图大法魔咒百科词典 (Bybilibili波西BrackRat)2、样例使用文字描述第一区►Prompt第一行(masterpiece:1.0),(bestquality:1.0),(ultrahighres:1.0),(8kresolution:1.0),(realistic:1.0),(ultradetailed1:0),(sharpfocus1:0),(RAWphoto:1.0)第二行1girl,......
  • 基于容器平台 ACK 快速搭建 Stable Diffusion
    作者:子白本文介绍如何在阿里云容器平台ACK上快速搭建一套可对外提供服务的StableDiffusion。CPU版本前提条件已创建Kubernetes托管版集群。具体操作,请参见创建Kubernetes托管版集群[1]。......
  • Adobe Photoshop 2023(MAC+Windows) +AI插件auto Photoshop stable diffusion plugin
    Adobe图像处理软件Photoshop2023正式版(24.1.1)2023年01月版发布。AdobePhotoshop2023破解版(简称PS)是一款全球流行的专业图像处理软件及照片和设计软件。AdobePhotoshop中文破解版是AdobeCreativeCloud创意云桌面程序中心的图形设计软件热门产品,它是平面设计领域和数......
  • 阿里云 AIGC 白嫖 FC 搭建 stable diffusion
    下午瞎逛在V站看到阿里在做推广,正好这几天在研究stable-diffusion,就进去看了看,活动地址:https://developer.aliyun.com/topic/aigc。主要就是阿里云的FC免费提供3个月的试用(注意,只有150元额度,所以重度使用可能一会就玩没了),可以快速搭建AiGC服务。安装注意阿里云官......
  • AtCoder Regular Contest 114 D Moving Pieces on Line
    洛谷传送门AtCoder传送门挺有意思的题。首先显然地,一个棋子不会走回头路。于是一个棋子沿着边走的效果就是区间异或。更进一步,设\(s_i\)为\(i-1\toi\)的边颜色与\(i\toi+1\)的边颜色是否相同(差分),相当于对于每个\(i\)都选择\(s_{a_i}\)和\(s_{x_i}\),将它们异或......
  • 猛读论文6 |【CVPR 2022】Camera-Conditioned Stable Feature Generation for Isolate
    用于孤立摄像机监督行人重识别的摄像机条件稳定特征生成动机常规ReID,对于一个ID,在不同摄像头拍摄的图片上提取跨相机视图不变特征而ISCS情况下,无法做到同一个ID采集到不同摄像头图片由于跨相机样本在人体Re-ID模型训练中起着重要作用,而在ISCS设置下不存在此类配对图像,因......
  • mac使用Stable Diffusion基础篇
    准备工作提前安装git.python等必要工具 1、git拉取WebUI仓库 gitclonehttps://github.com/AUTOMATIC1111/stable-diffusion-webui2、下载StableDiffusionModelshttps://huggingface.co/CompVis/stable-diffusion-v-1-4-original  3、启动下载完成后,把下载的s......
  • 最新版本 Stable Diffusion 开源 AI 绘画工具之中文自动提词篇
    目录......
  • 玩转AIGC,5分钟 Serverless 部署 Stable Diffustion 服务
    有没有一种可能,其实你早就在AIGC了?阿里云将提供免费Serverless函数计算产品资源,邀请你,体验一把AIGC级的毕加索、达芬奇、梵高等大师作画的快感。下面请尽情发挥你的想象空间!!双重奖品设置,完成体验场景可得社区1000积分兑换奖品,还可参加AI生成图像比赛赢取Airpods、500元猫超卡......