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>1:
disp(name+',耗时:'+str(t2) +'小时')
elif t1>1:
disp(name+',耗时:'+str(t1) +'分钟')
elif t<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>1:
disp(name+',耗时:'+str(t2) +'小时')
elif t1>1:
disp(name+',耗时:'+str(t1) +'分钟')
elif t<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>