Python 算法交易实验72 QTV200第一步: 获取原始数据并存入队列

2024-06-24 01:04
文章标签 python 算法 实验 队列 获取 交易 第一步 72 存入 原始数据 qtv200

本文主要是介绍Python 算法交易实验72 QTV200第一步: 获取原始数据并存入队列，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

说明

最近的数据流往前进了一步，我觉得基本可以开始同步的推进QTV200了。上次规划了整体的数据流，现在开始第一步。

内容

1 结构位置

这是上次的总体图：
在这里插入图片描述
以下是这次要实现的一小部分：

在这里插入图片描述
从结构上，这个是整体数据流的起点，系统因为这些不断 运行的数据才开始“动”了起来，可以称为源点。

2 规范与约束

源点是基于每分钟的节拍从外界读取数据，这部分目前我没用用付费接口(数据的需求量很小)，所以基于自律(类似与吃自助餐)的原则，增加一些规范与约束。

  • 我获取的数据不会多，可以约束在60个ETF之内。
  • 每次请求只会查询当前时刻的前10分钟数据(数据少),每只ETF一天最多有60610 ~ 3600条数据
  • 每次请求10条的目的是为了防止某个时隙程序中断或者失效，通过冗余的数据可以在10分钟之内的终端内无缝恢复（从这角度，用某个云服务器做这件事比较合适,下一版考虑)
  • 任务按照秒进行划分，每秒最多提交6只ETF的请求，数据请求总量为60条
  • 周末一定不会发起查询请求

这样可以确保非必要不请求数据，即使请求数据，请求也被均匀分摊，每次的请求量非常之小（环保）

3 工具与方法

通过FLask-APS执行秒级的任务调度，通过Flask-Celery实现各ETF的异步抓取，确保时效的同时，减少CPU开销。(同步方式会独占一个核，很浪费的)。

有的时候倒也不纯粹是为了节约这点计算成本，而是总体成本。设想，一开始只跟踪3~4个ETF，同步状态下并发，可能抢占4个核一小会，还不至于出现卡顿(主机有32核)。但是如果跟踪60个ETF，那么整个机器就会因为这个原因处于卡顿状态，那就真的很没必要。

即使是现阶段，QTV102与Mongo通信的时候更新少量数据,但是是同步状态的，都让我的CPU负载处于一个很奇怪的状态。
在这里插入图片描述
虽然看着很满，其实我知道很多是处于浪费的状态的。

所以，如果用合适的方法进行调度，那么即使是60个ETF，甚至是600个ETF可能单核都足够了。这种效率的提升是很夸张的。目前在IO密集处理这块可以做到的提升最大。未来QTV200实现后，应该会把QTV102整个卸载掉。

3.1 worker

对于每一个ETF来说，处理的流程是相同的。所以，可以先做一个worker,然后调用的时候按不通的ETF代码进行参数化就可以了。

etf_crawl_worker.py

# 0 记录日志
import logging
from logging.handlers import RotatingFileHandlerlogger = logging.getLogger('MyLogger')
handler = RotatingFileHandler('/var/log/workers.log', maxBytes=1024*1024*100, backupCount=5)
logger.addHandler(handler)
logger.setLevel(logging.INFO)# 1 允许传入一个参数
import argparse
def get_arg():parser = argparse.ArgumentParser(description='Customized Arguments')
#     parser.add_argument('-p','--pkl', default='Meta')# 制程名parser.add_argument('--etf_code')parser.add_argument('--assigned_work_dt')# 准备解析参数args = parser.parse_args()res_dict = {}res_dict['etf_code'] = args.etf_coderes_dict['assigned_work_dt'] = args.assigned_work_dtreturn res_dictarg_dict = get_arg()etf_code = arg_dict['etf_code']
assigned_work_dt = arg_dict.get('assigned_work_dt')from Basefuncs import * 
import time# 2 判断是否是可执行时间
if assigned_work_dt is None:ts = time.time()
else:if assigned_work_dt.strip() == '':ts = time.time()else:ts = inverse_time_str(assigned_work_dt)cur_dt_str = get_time_str1(ts)
cur_time = cur_dt_str.split()[-1]morning_start = '09:25:00'
morning_end = '11:41:00'
afternoon_start = '12:55:00'
afternoon_end = '15:11:00'is_moring_work = False 
is_afternoon_work = False
if cur_time >= morning_start and cur_time < morning_end:is_moring_work = True 
if cur_time >= afternoon_start and cur_time < afternoon_end:is_afternoon_work = True is_work_time = is_moring_work or is_afternoon_workif is_work_time:start_dt = get_time_str1( (ts//60)*60 - 600 )end_dt = get_time_str1( ts + 60 )# 目标队列设置qm = QManager(redis_agent_host = 'http://192.168.0.4:24118/',redis_connection_hash = None)# qm.info()# 3 执行# etf_code = '510300'import akshare as ak para_dict  ={}para_dict['symbol'] = etf_codepara_dict['period'] = "1"para_dict['adjust'] = ''para_dict['start_date'] = start_dtpara_dict['end_date'] = end_dt# 如果时间段不对，那么就是空df = ak.fund_etf_hist_min_em(**para_dict)# 是否获取到了数据is_query_data = True if len(df) else False if is_query_data:# ak的变量字典映射ak_dict = {}ak_dict['时间'] = 'data_dt'ak_dict['开盘'] = 'open'ak_dict['收盘'] = 'close'ak_dict['最高'] = 'high'ak_dict['最低'] = 'low'ak_dict['成交量'] = 'vol'ak_dict['成交额'] = 'amt'keep_cols = ['data_dt','open','close','high','low','vol','amt']cols = list(df.columns)new_cols = [ak_dict.get(x) or x for x in cols ]df.columns = new_colsdf1 = df[keep_cols]df1['data_source'] = 'AK'df1['code'] = etf_codedf1['market'] = 'SH'df1['rec_id'] = df1['data_source'] + '_' + df1['market'] + '_' + df1['code'].apply(str) \+ '_' + df1['data_dt']# 调整股和手vol_approximal = df1['amt'] / df1['close']maybe_wrong = (vol_approximal / df1['vol']) > 10if maybe_wrong.sum() > 0:df1['vol'] = df1['vol'] * 100stream_name = 'YOURS.stream_in'# 写入结果队列data_listofdict = df1.to_dict(orient='records')resp = qm.parrallel_write_msg(stream_name, data_listofdict)logger.info('%s %s 【%s】' % (cur_dt_str,'etf_crawl_worker',resp['msg'] ))   else:logger.info('%s %s 【未获取到数据】' % (cur_dt_str,'etf_crawl_worker'))   else:logger.info('%s %s 【不在工作时间】' % (cur_dt_str,'etf_crawl_worker'))

worker分为几部分：

  • 1 设定使用rotate日志，记录每次执行的效果
  • 2 get_arg 获取调用时传入的关键字参数
  • 3 判断是否在工作时间。默认情况，使用当前时间；也可接受使用指定的时间
  • 4 如果是在工作时间，那么推算对应时间的前10分钟和后1分钟，作为参数发起请求
  • 5 获取数据后，还有一个判断交易量是手还是股的小逻辑
  • 6 处理完成后，推入队列，然后记录日志，worker执行完毕

两种调用方法：

  • 1 实时获取
python3 etf_crawl_worker.py --etf_code=510300
  • 2 指定历史时间的获取
python3 etf_crawl_worker.py --etf_code=510300 --assigned_work_dt='2024-06-21 10:00:00'

获取日志查看，第三条数据是因为指定了工作时间。

/var/log/workers.log

└─ $ cat workers.log
2024-06-23 20:27:07 etf_crawl_worker 【不在工作时间】
2024-06-23 20:40:41 etf_crawl_worker 【不在工作时间】
2024-06-23 20:57:48 etf_crawl_worker 【ok,add 12 of 12  messages】

3.2 shell

exe_etf_crawl_worker.sh
chmod +x exe_etf_crawl_worker.sh

本来想让脚本可以接受第二个参数的，但是似乎隔空传带空格的参数很麻烦，另外实时调用(脚本)的时候没有必要再筛选时间了。

#!/bin/bash# 记录
# sh /home/test_exe.sh com_info_change_pattern running# 有些情况需要把source替换为 .
# . /root/anaconda3/etc/profile.d/conda.sh
# 激活 base 环境（或你创建的特定环境）
source /root/miniconda3/etc/profile.d/conda.sh#conda init
conda activate basecd  /home/workers && python3 etf_crawl_worker.py --etf_code=$1

3.3 flask_celery

将flask_celery升级为可执行脚本的版本

In [1]: import requests as req...: param_dict = {'the_cmd': 'bash /home/test_exe.sh'}...: resp = req.post('http://127.0.0.1:24104/exe_sh/',json = param_dict )In [2]: resp
Out[2]: <Response [200]>└─ $ cat tem.log
2024-06-23 22:16:25 - 脚本执行

3.4 将任务发布为flask_aps任务

任务参数如下

# 任务6：执行脚本-qtv200 510300 get 
task006 = {}
task006['machine'] = 'm4'
task006['task_id'] = 'task006'
task006['description'] = '执行脚本,在周一到周五，上午9点到下午4点执行，获取510300的数据。在秒0执行'
task006['pid'] = '.'.join([task006['machine'],task006['task_id']  ])
task006['job_name'] = 'make_a_request' # 这个是对flask-aps来说的
task006['set_to_status'] = 'running'
task006['running_status'] = ''
task006['start_dt'] = '2024-05-01 00:00:00'
task006['end_dt'] = '2099-06-01 00:00:00'
task006['task_kwargs'] = {'para_dict': {'url':'http://172.17.0.1:24104/exe_sh/','json_data':{'the_cmd': 'bash /home/exe_etf_crawl_worker.sh 510300'}}}
task006['interval_para'] ={'second':'0','day_of_week':'0-4','hour':'9-16'}
task006 = TaskTable(**task006)
task006.save()

ok，明天等着看吧

In [11]: the_task_obj = TaskTable.objects(machine='m4',task_id ='task006').first()...: exe_a_task(the_task_obj)
set to status running  current state  init
Publish a task
Out[11]: 1

这篇关于Python 算法交易实验72 QTV200第一步: 获取原始数据并存入队列的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1088761。

相关文章

Python正则表达式语法及re模块中的常用函数详解

Python正则表达式语法及re模块中的常用函数详解

《Python正则表达式语法及re模块中的常用函数详解》这篇文章主要给大家介绍了关于Python正则表达式语法及re模块中常用函数的相关资料,正则表达式是一种强大的字符串处理工具,可以用于匹配、切分、... 目录概念、作用和步骤语法re模块中的常用函数总结 概念、作用和步骤概念： 本身也是一个字符串，其中
阅读更多...
Python使用getopt处理命令行参数示例解析(最佳实践)

Python使用getopt处理命令行参数示例解析(最佳实践)

《Python使用getopt处理命令行参数示例解析(最佳实践)》getopt模块是Python标准库中一个简单但强大的命令行参数处理工具,它特别适合那些需要快速实现基本命令行参数解析的场景,或者需要... 目录为什么需要处理命令行参数？getopt模块基础实际应用示例与其他参数处理方式的比较常见问http
阅读更多...
python实现svg图片转换为png和gif

python实现svg图片转换为png和gif

《python实现svg图片转换为png和gif》这篇文章主要为大家详细介绍了python如何实现将svg图片格式转换为png和gif,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录python实现svg图片转换为png和gifpython实现图片格式之间的相互转换延展：基于Py
阅读更多...
Python中的getopt模块用法小结

Python中的getopt模块用法小结

《Python中的getopt模块用法小结》getopt.getopt()函数是Python中用于解析命令行参数的标准库函数,该函数可以从命令行中提取选项和参数,并对它们进行处理,本文详细介绍了Pyt... 目录getopt模块介绍getopt.getopt函数的介绍getopt模块的常用用法getopt模
阅读更多...
Python利用ElementTree实现快速解析XML文件

Python利用ElementTree实现快速解析XML文件

《Python利用ElementTree实现快速解析XML文件》ElementTree是Python标准库的一部分,而且是Python标准库中用于解析和操作XML数据的模块,下面小编就来和大家详细讲讲... 目录一、XML文件解析到底有多重要二、ElementTree快速入门1. 加载XML的两种方式2.
阅读更多...
Python如何精准判断某个进程是否在运行

Python如何精准判断某个进程是否在运行

《Python如何精准判断某个进程是否在运行》这篇文章主要为大家详细介绍了Python如何精准判断某个进程是否在运行,本文为大家整理了3种方法并进行了对比,有需要的小伙伴可以跟随小编一起学习一下... 目录一、为什么需要判断进程是否存在二、方法1：用psutil库（推荐）三、方法2：用os.system调用
阅读更多...
Java的栈与队列实现代码解析

Java的栈与队列实现代码解析

《Java的栈与队列实现代码解析》栈是常见的线性数据结构,栈的特点是以先进后出的形式,后进先出,先进后出,分为栈底和栈顶,栈应用于内存的分配,表达式求值,存储临时的数据和方法的调用等,本文给大家介绍J... 目录栈的概念（Stack）栈的实现代码队列（Queue）模拟实现队列(双链表实现)循环队列（循环数组
阅读更多...
使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)

使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)

《使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)》PPT是一种高效的信息展示工具,广泛应用于教育、商务和设计等多个领域,PPT文档中常常包含丰富的图片内容,这些图片不仅提升了... 目录一、引言二、环境与工具三、python 提取PPT背景图片3.1 提取幻灯片背景图片3.2 提取
阅读更多...
Python实现图片分割的多种方法总结

Python实现图片分割的多种方法总结

《Python实现图片分割的多种方法总结》图片分割是图像处理中的一个重要任务,它的目标是将图像划分为多个区域或者对象,本文为大家整理了一些常用的分割方法,大家可以根据需求自行选择... 目录1. 基于传统图像处理的分割方法(1) 使用固定阈值分割图片(2) 自适应阈值分割(3) 使用图像边缘检测分割(4)
阅读更多...
C# foreach 循环中获取索引的实现方式

C# foreach 循环中获取索引的实现方式

《C#foreach循环中获取索引的实现方式》：本文主要介绍C#foreach循环中获取索引的实现方式,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录一、手动维护索引变量二、LINQ Select + 元组解构三、扩展方法封装索引四、使用 for 循环替代
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2