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 FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

《PythonFastAPI+Celery+RabbitMQ实现分布式图片水印处理系统》这篇文章主要为大家详细介绍了PythonFastAPI如何结合Celery以及RabbitMQ实现简单的分布式... 实现思路FastAPI 服务器Celery 任务队列RabbitMQ 作为消息代理定时任务处理完整
阅读更多...
Python Websockets库的使用指南

Python Websockets库的使用指南

《PythonWebsockets库的使用指南》pythonwebsockets库是一个用于创建WebSocket服务器和客户端的Python库,它提供了一种简单的方式来实现实时通信,支持异步和同步... 目录一、WebSocket 简介二、python 的 websockets 库安装三、完整代码示例1.
阅读更多...
揭秘Python Socket网络编程的7种硬核用法

揭秘Python Socket网络编程的7种硬核用法

《揭秘PythonSocket网络编程的7种硬核用法》Socket不仅能做聊天室,还能干一大堆硬核操作,这篇文章就带大家看看Python网络编程的7种超实用玩法,感兴趣的小伙伴可以跟随小编一起... 目录1.端口扫描器：探测开放端口2.简易 HTTP 服务器：10 秒搭个网页3.局域网游戏：多人联机对战4.
阅读更多...
使用Python实现快速搭建本地HTTP服务器

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》：本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.
阅读更多...
Python使用自带的base64库进行base64编码和解码

Python使用自带的base64库进行base64编码和解码

《Python使用自带的base64库进行base64编码和解码》在Python中,处理数据的编码和解码是数据传输和存储中非常普遍的需求,其中,Base64是一种常用的编码方案,本文我将详细介绍如何使... 目录引言使用python的base64库进行编码和解码编码函数解码函数Base64编码的应用场景注意
阅读更多...
Python基于wxPython和FFmpeg开发一个视频标签工具

Python基于wxPython和FFmpeg开发一个视频标签工具

《Python基于wxPython和FFmpeg开发一个视频标签工具》在当今数字媒体时代,视频内容的管理和标记变得越来越重要,无论是研究人员需要对实验视频进行时间点标记,还是个人用户希望对家庭视频进行... 目录引言1. 应用概述2. 技术栈分析2.1 核心库和模块2.2 wxpython作为GUI选择的优
阅读更多...
Python如何使用__slots__实现节省内存和性能优化

Python如何使用__slots__实现节省内存和性能优化

《Python如何使用__slots__实现节省内存和性能优化》你有想过,一个小小的__slots__能让你的Python类内存消耗直接减半吗,没错,今天咱们要聊的就是这个让人眼前一亮的技巧,感兴趣的... 目录背景：内存吃得满满的类__slots__：你的内存管理小助手举个大概的例子：看看效果如何？1.
阅读更多...
Python+PyQt5实现多屏幕协同播放功能

Python+PyQt5实现多屏幕协同播放功能

《Python+PyQt5实现多屏幕协同播放功能》在现代会议展示、数字广告、展览展示等场景中,多屏幕协同播放已成为刚需,下面我们就来看看如何利用Python和PyQt5开发一套功能强大的跨屏播控系统吧... 目录一、项目概述：突破传统播放限制二、核心技术解析2.1 多屏管理机制2.2 播放引擎设计2.3 专
阅读更多...
Python中随机休眠技术原理与应用详解

Python中随机休眠技术原理与应用详解

《Python中随机休眠技术原理与应用详解》在编程中,让程序暂停执行特定时间是常见需求,当需要引入不确定性时,随机休眠就成为关键技巧,下面我们就来看看Python中随机休眠技术的具体实现与应用吧... 目录引言一、实现原理与基础方法1.1 核心函数解析1.2 基础实现模板1.3 整数版实现二、典型应用场景2
阅读更多...
Python实现无痛修改第三方库源码的方法详解

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2