Python一些可能用的到的函数系列130 UCS-Time Brick

说明

UCS对象是基于GFGoLite进行封装，且侧重于实现UCS规范。

内容

1 函数

我发现pydantic真是一个特别好用的东西，可以确保在数据传递时的可靠，以及对某个数据模型的描述。

以下，UCS给出了id、time相关的brick映射，并给出了时间到字符的转换。其方便之处在于，本地几乎不需要管怎么实现的，只要有服务支持即可。效率也足够高。

from typing import List, Optional
from pydantic import BaseModelimport requests as req 
class UCS(BaseModel):__version__ =1.1gfgo_lite_server: str = 'http://172.17.0.1:24090/'def get_brick_name(self, some_id = None):some_dict = {}some_dict['rec_id'] = some_idurl = self.gfgo_lite_server + 'get_brick_name/'res = req.post(url, json = some_dict).json()return res def get_brick_name_s(self, some_id_list = None):some_dict = {}some_dict['rec_id_list'] = some_id_listurl = self.gfgo_lite_server + 'get_brick_name_s/'res = req.post(url, json = some_dict).json()return res def gfgo(self, kwargs = None, pack_func = None):url = self.gfgo_lite_server + 'gfgo/'return req.post(url, json = {'kwargs':kwargs,'pack_func':pack_func}).json()def dt_str2num(self, some_dt_str = None):some_dict = {}some_dict['some_dt_str'] = some_dt_strurl = self.gfgo_lite_server + 'str2num/'res = req.post(url, json = some_dict).json()return  resdef dt_str2num_s(self, some_dt_str_list = None):some_dict = {}some_dict['some_dt_str_list'] = some_dt_str_listurl = self.gfgo_lite_server + 'str2num_s/'res = req.post(url, json = some_dict).json()return  res# 时间名称def get_time_brick_name(self, dt_str_or_ts = None):some_dict = {}some_dict['dt_str_or_ts'] = dt_str_or_tsurl = self.gfgo_lite_server + 'get_time_brick_name/'res = req.post(url, json = some_dict).json()return res def get_time_brick_name_s(self, dt_str_or_ts_list = None):some_dict = {}some_dict['dt_str_or_ts_list'] = dt_str_or_ts_listurl = self.gfgo_lite_server + 'get_time_brick_name_s/'res = req.post(url, json = some_dict).json()return res # 1.1 >>>def get_brick_list(self, start_brick_name = None, end_brick_name = None):some_dict = {}some_dict['start_brick_name'] = start_brick_namesome_dict['end_brick_name'] = end_brick_nameurl = self.gfgo_lite_server + 'get_brick_list/'res = req.post(url, json = some_dict).json()return res def get_brick_bounds(self, brick_name = None):some_dict = {}some_dict['brick_name'] = brick_nameurl = self.gfgo_lite_server + 'get_brick_bounds/'res = req.post(url, json = some_dict).json()return res # 1.2 >>> 时间# get_time_brick_bounds_s - 待def get_time_brick_bounds_s(self, brick_name_list = None, char_or_num = 'char'):some_dict = {}some_dict['brick_name_list'] = brick_name_listsome_dict['char_or_num'] = char_or_numurl = self.gfgo_lite_server + 'get_time_brick_bounds_s/'res = req.post(url, json = some_dict).json()return res # get_time_brick_listdef get_time_brick_list(self,start_brick_name = None, end_brick_name = None):some_dict = {}some_dict['start_brick_name'] = start_brick_namesome_dict['end_brick_name'] = end_brick_nameurl = self.gfgo_lite_server + 'get_time_brick_list/'res = req.post(url, json = some_dict).json()return res 

2 应用

2.1 时间转换

ucs = UCS(gfgo_lite_server = 'http://xxx:24090/')
ucs.dt_str2num('2024-01-31 11:11:11')
1706670671ucs.dt_str2num_s(['2024-01-31 11:11:11','2024-02-01 11:11:11' ])
[1706670671, 1706757071]

反函数可以用get_time_str1

In [19]: get_time_str1(1706670671)
Out[19]: '2024-01-31 11:11:11'

2.2 UCS-time

一个简单的应用如下：我需要制定一个静态数据同步计划，计划需要随着时间轴，按照时间块熟顺序向前推进。具体来说，这个对于数据库迁移(合并)与回测都有关系。

tbrick1 = ucs.get_time_brick_name('2024-01-31 11:11:11')
'2024.01.31.11'
tbrick2 = ucs.get_time_brick_name('2024-02-10 11:11:11')
'2024.02.10.11'tbrick_list = ucs.get_time_brick_list(start_brick_name = tbrick1, end_brick_name = tbrick2)
['2024.01.31.11', '2024.01.31.12', '2024.01.31.13', ...]ucs.get_time_brick_bounds_s(tbrick_list[:3])
[['2024-01-31 11:00:00', '2024-01-31 12:00:00'],['2024-01-31 12:00:00', '2024-01-31 13:00:00'],['2024-01-31 13:00:00', '2024-01-31 14:00:00']]

btw, 在探索完UCS-time之后，我发现大约因为时间观念比较深入人心，其实是可以更自由的转换的，

# 将一般字符串转为UCS 名称
def dt_str2ucs_blockname(some_dt_str):some_dt_str1   =some_dt_str.replace('-','.').replace(' ','.').replace(':','.')return '.'.join(some_dt_str1.split('.')[:4])'''
dt_str2ucs_blockname('2024-06-24 09:30:00')
'2024.06.24.09'
'''

这些函数也被纳入基础函数，有时候可以起到辅助的作用,特别在使用GFGolite不那么方便的时候。

一个更具体的场景，我觉得是具有通用性的，描述如下：

• 1 数据没有顺序id，但是具备入库时间(create_time)
• 2 数据入库时间是不连续的，有一些时间段密集，更多的时间可能空闲，但也并不是一条没有。
• 3 每次执行时，worker总是能够有效推进一个brick

具体做法：

p01: 将需要的时间块按序持久化为本地文件，并将对应的bounds作为字典

p02: 获取当前时间的时间块

# 将标准时间字符串转为brick
def timestr2timeblock(some_time_str = None):x1 = some_time_str.replace('-','.').replace(' ','.').replace(':','.')return '.'.join(x1.split('.')[:4])the_time_now = get_time_str1()
the_time_brick = timestr2timeblock(the_time_now)

p03:从gb中获取上一次处理过的最新brick，并锁定在时间轴的位置。

worker_buffer_space = 'sp_worker.general'
tier1 = 'some_tier'
tier2 = 'ucs_time_brick_ordered.sniffer'
prefix = '.'.join([worker_buffer_space,tier1,tier2]) +'.'# ==========================  Load
time_brick_list = from_pickle('time_brick_list')
time_bounds_dict = from_pickle('time_bounds_dict')
gb = GlobalBuffer()
watch_brick_pos = time_brick_list.index(the_time_brick)

p04:按照已处理之后，当前观察时间之前的原则，选出available_time_brick_list

latest_data_brick = gb.getx(prefix +'last_brick_handled')
if latest_data_brick is None:start_pos = 0
else:start_pos =  time_brick_list.index(latest_data_brick)available_time_brick_list = time_brick_list[start_pos+1:watch_brick_pos]

p05:按顺序遍历时间块，直到处理到一个有效的brick才中断本次循环

# cur_data_brick = available_time_brick_list[0]
for cur_data_brick in available_time_brick_list:print(cur_data_brick)cur_bound = time_bounds_dict[cur_data_brick]print(cur_bound)resp = 【获取待处理数据】if len(resp):res_df = pd.DataFrame(resp, columns = keep_cols)if ok_to_handle:【将结果分批删除】listofdict2 = slice_list_by_batch2(xxx.to_dict(orient='records'), 1000)for some_listofdict in listofdict2:【更新已处理块信息】gb.setx(prefix +'last_brick_handled',cur_data_brick, persist=True)break

通过循环或者定时任务唤起这样的worker就可以。