既要坚定锻炼成熟架构的道路，也要在合理的范围内重塑设计
计算时序数据的特征，少不了“Rolling”类的操作。过去，直接采用pandas进行rolling，效率很不错，但是在实战应用时不太行。
反思下来：离线的操作拓展困难，很多都是一次性的
后来开发了ADBS，通过Mongo和Redis，在数据的持久化和吞吐上是没问题的。但是，对于全量的历史回滚计算遇到了问题。
ADBS的架构将问题简化到了一个时隙，大大简化了逻辑，开发完成也即生产完成，这点很好；但是在大量的全量计算中，Mongo库还是碰到了“大频次”访问的问题。目前我关注的时隙大约是60万，也就是说各Worker会频繁的发出数据查询请求。最终优化下来，处理的时间大约要1.5天。实时计算方面倒没什么问题，(每分钟计算300时隙 – 这不就是 TPS = 5吗?）
之前纠结的点在于使用硬件去堆高能力，而不去做设计上优化。但今天转念一想，不能轴啊，毕竟可以用很小的代价来实现的。
可使用内存服务的点在于：
TPS = 5 ，那么一天计算约40万次计算；因为我把时隙定在分钟，所以应该是TPM=300。
从吞吐的指标上看，至少要能达到TPS = 100或者 TPM = 6000这个水平。这样处理速度快20倍，不到1个小时就能跑完全量。
构造一个服务，服务的主要目的是将这种滚动式的数据查询从数据库中挪到内存中
服务的构造上，在这个场景用Tornado是最合适的，不过我不想花时间去搞；还是要用Flask搭建，未来有时间了，还是应该把Tornado的模板搞一搞，又轻又快。
服务能够提速的原因在于：
顺带的能稍微节约一下硬盘的耐久度消耗（虽然有时候我也暗搓搓的希望硬盘用坏了可以买新的，哈哈）。
服务最大的开销可能就是将结果进行序列化，然后在网络(局网)中传输了。
其实如果是非滚动的计算，当前的架构是完全可以满足的，所以新的服务应该只是这个特殊场景的补充，用最小化的功能来进行设计。
按 random_seed ，从某个库拖取区间数据
接口的入参需要指定：
接口服务在拉取数据后存入服务的内存变量（字典）：
在不使用时，通过重启容器来释放内存
和原来查表同样的入参
一般来说，通过一组变量作为联合主键来进行数据的筛选，然后通过时隙的比较来进行取数，之后将结果作为listofdict返回。
采用了Tornado做网络服务，的确是比Flask强很多。
单核情况下，T的处理大约是120ms~150ms, F的处理 一般在200ms。所以T的TPS是7，而F的TPS是5。
在概念上，我有两个认知盲点或者错误：
取数的时候，如果直接从Mongo取，那么需要数据库进行查询，这个会耗用CPU；当采用内存处理时，从python的对象转为json传输，也要耗用CPU。可能后者的耗用还会大一些。
因为现在都是在使用MA进行数据操作，MA为了标准化操作，也会进行转换，所以我觉得效率都消耗在这上面。
速度的确提高了很多，本以为这就是这个故事的结果，没想到又反转了。
tornado在多核运行时出现了内存暴增的问题。我估计在多核状态下，可能还会有很多进程会被随机创建，然后每个进程会去获取静态数据（是的，肯定不是十个核拿十次，而是有多少进程多少次）。所以多核的时候，内存方案会有问题。我想以前碰到的一些（其他人）的工程上内存爆掉的问题，可能有一部分是类似这样的。python因为天生是单进程的，所以不太会体会到。
我觉得，如果要多核操作，一定还是要面向数据库的，还是转向redis找解决方案
擦，突然发现其实我已经有了答案，灯下黑啊。
我现在往Redis的Stream不就是存取数据，然后分发执行吗？效率很高啊。只是有一个地方需要改变，就是消息的id用我的时隙号替代就可以了…
甚至再往前一步，所有的实时数据可以一直往这个队列加啊，之前都是什么鬼方案，哈哈。
在python里使用redis操作，原来我就是没有使用id这个关键字。
所以只要在现有的redis agent里加一个接口就行(10分钟都不用），完全不影响其他的操作。这样，就把这个问题重新简化为了IO问题。之前也看过，这60多万的内存占用应该也就200多M。
将mongo中的数据都搬到redis(60多万条数据，耗时7分钟左右)
100%|██████████| 115/115 [06:49<00:00, 3.57s/it]
from configs_base import redis_agent_host,project_name, worker01_config,cur_w
tier1 = 'MyQuantBaseStep1Signals'
tier2 = 'step1_mongo_in'
min_slot = cur_w.minmax(tier1 = tier1, tier2 = tier2, minmax='min', attrname='data_slot')['data']
max_slot = cur_w.minmax(tier1 = tier1, tier2 = tier2, minmax='max', attrname='data_slot')['data']
print(min_slot)
print(max_slot)
slot_tuple_list = slice_list_by_batch1(int(min_slot),int(max_slot)+1,50000)
redis_q_name = 'BUFF.%s.%s' %(tier1, tier2)
import tqdm
for tem_tuple in tqdm.tqdm(slot_tuple_list):
min_time_slot, max_time_slot = tem_tuple
recs = cur_w.query_recs(tier1 = tier1, tier2 = tier2,
filter_dict = {'$and':[{'market':market,
'code':code, 'data_slot':{'$gte':min_time_slot ,'$lt':max_time_slot}}]},
limits= (max_time_slot - min_time_slot) * 10,silent=True
)['data']
recs_df = pd.DataFrame(recs)
recs_df['_data_source_ranking'] = recs_df['data_source'].map(data_source_ranking).fillna(999)
recs_df1 = recs_df.sort_values(['_data_source_ranking']).drop_duplicates(['market','code','data_slot'])
tem_msg_id_list
keep_cols = [x for x in recs_df1.columns if not x.startswith('_')]
recs_df1 = recs_df1.sort_values(['data_slot'])
tem_msg_dict_list = recs_df1[keep_cols].to_dict(orient='records')
tem_msg_id_list = list(recs_df1['data_slot'].apply(int))
req.post(redis_agent_host + 'batch_add_msg_with_id/',
json={'stream_name':redis_q_name,
'msg_dict_list':tem_msg_dict_list,
'msg_id_list': tem_msg_id_list,
'maxlen':10000000}).json()
Worker从Redis取数
import redis
lq = redis.Redis()
def flat_kvlist2dict(some_list):
tem_dict = {}
for i in range(0,len(some_list),2):
k = some_list[i]
v = some_list[i+1]
tem_dict[k] = v
return tem_dict
%%timeit
stream_name = redis_q_name
start_id = 28001000
end_id = 28023010
count = 100000
cmd = 'xrange %s %s %s count %s' %(stream_name,start_id,end_id,count)
print(cmd)
res = lq.execute_command(cmd)
res_df = pd.DataFrame(res)
msg_body_df = pd.DataFrame(res_df[1].apply(flat_kvlist2dict).tolist())
每次的取数大约20多毫秒(TPS ~ 50),看起来不再需要考虑序列化的问题（一来一去，每次全量减少120万次的json序列化工作）
再之后修改Worker的取数方式就好了。