wtpy使用手册

wtpy开源项目的使用


k线数据合成函数编写

<ol> <li> <p>从tb导出的rb一分钟历史数据 <a href="https://pan.baidu.com/s/1FnTIValbeb2UU30g3uRMTw?pwd=odzz" title="下载地址">下载地址</a></p> </li> <li> <p>代码如下:</p> <pre><code class="language-python"># -*- coding: utf-8 -*- import time import http.client import os import numpy as np import pandas as pd import datetime import pickle import threading import re import re import pandas as pd import time from wtpy.wrapper import WtDataHelper from wtpy.WtCoreDefs import WTSBarStruct, WTSTickStruct from tqdm import tqdm import numpy as np def disp(strG='',strTime=''): t=time.localtime(time.time()) tS=time.strftime("%Y-%m-%d %H:%M:%S",t) if strTime=='': # 默认打印 print(tS+' '+str(strG)) elif strTime=='日期': return time.strftime("%Y-%m-%d",t) elif strTime=='时间': return time.strftime("%H:%M:%S",t) elif strTime=='日期时间': return tS elif strTime=='日期_': return time.strftime("%Y%m%d",t) elif strTime=='日期_n': return int(time.strftime("%Y%m%d",t)) elif strTime=='时间_': return time.strftime("%H%M%S",t) elif strTime=='时间_n': return int(time.strftime("%H%M%S",t)) elif strTime=='日期时间_': return time.strftime("%Y%m%d%H%M%S",t) else: print('disp函数参数错误!!!!!!!!!!!!!!!!!') class tt: def __init__(self): self.start_time=time.time() def get(self): t=round(time.time()-self.start_time,2) return t def get_p(self,name=''): name=str(name) t=self.get() t1=round(t/60,2) t2=round(t/60/60,2) if t2&gt;1: disp(name+',耗时:'+str(t2) +'小时') elif t1&gt;1: disp(name+',耗时:'+str(t1) +'分钟') elif t&lt;1: disp(name+',耗时:'+str(t*1000)+'毫秒') else: disp(name+',耗时:'+str(t)+'秒') def get_pp(self,name=''): name=str(name) t=self.get() t1=round(t/60,2) t2=round(t/60/60,2) if t2&gt;1: disp(name+',耗时:'+str(t2) +'小时') elif t1&gt;1: disp(name+',耗时:'+str(t1) +'分钟') elif t&lt;1: disp(name+',耗时:'+str(t*1000)+'毫秒') else: disp(name+',耗时:'+str(t)+'秒') return t1 t=tt() f = open('rb000.csv','r') file_data = f.readlines() #读取所有行 f.close() file_data1=[re.split(',',i)for i in file_data] names =['datetime','open','high','low','close','vol','hold'] df=pd.DataFrame(file_data1[1:],columns=names) df['open']=df['open'].astype("float") df['high']=df['high'].astype("float") df['low']=df['low'].astype("float") df['close']=df['close'].astype("float") df['vol']=df['vol'].astype("int64") df['hold']=df['hold'].astype("int64") df['date']=df['datetime'].apply(lambda x:float(x[:10].replace('/',''))).astype("int64") df['time']=df['datetime'].apply(lambda x:float(x[11:].replace(':',''))).astype("int64") df['datetime']=df['datetime'].apply(lambda x:x.replace('/','-') ) df['time'] =(df['date']-19900000)*10**4 + df['time'] t.get_p('读取数据耗时1 ') t=tt() df['timestamp']=pd.to_datetime(df['datetime'])+ pd.Timedelta('1 min') # 偏离时间 df.set_index('timestamp', inplace=True) # on = 'timestamp', columns_list = ['open', 'high', 'low', 'close', 'vol', 'hold'] fun = ['first', 'max', 'min', 'last', 'sum','last'] data_interval = df.resample(rule='5min', label='right', closed='right').agg(dict(zip(columns_list,fun))) data_interval=data_interval[ ~np.isnan (data_interval['open'])] t.get_p('方法1 ') t=tt() rule_cycle = '5min' # cycle_df = pd.DataFrame() # cycle_df['open'] = df['open'].resample(rule=rule_cycle, label='right', closed='right').first() # cycle_df['close'] = df['close'].resample(rule=rule_cycle,label='right', closed='right').last() # # # 开、高、低的价格,成交量 cycle_df['high'] = df['high'].resample(rule=rule_cycle, label='right', closed='right').max() # cycle_df['low'] = df['low'].resample(rule=rule_cycle, label='right', closed='right').min() # cycle_df['vol'] = df['vol'].resample(rule=rule_cycle, label='right', closed='right').sum() # cycle_df['hold'] = df['hold'].resample(rule=rule_cycle, label='right', closed='right').last() cycle_df=cycle_df[ ~np.isnan (cycle_df['open'])] t.get_p('方法2') t=tt() </code></pre> </li> <li> <p>运行结果:</p> <p><img src="https://www.showdoc.com.cn/server/api/attachment/visitFile?sign=47cb25f34d1f972d5890950f5be5bb23" alt="" /></p> <p>2022-06-24 13:00:51 读取数据耗时1 ,耗时:4.95秒 2022-06-24 13:00:51 方法1 ,耗时:680.0毫秒 2022-06-24 13:00:54 方法2,耗时:2.65秒</p> </li> </ol>

页面列表

ITEM_HTML