-
Notifications
You must be signed in to change notification settings - Fork 43
/
Copy pathintf_test.py
101 lines (91 loc) · 3.72 KB
/
intf_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
#-*- coding:utf-8 -*-
import hashlib
import datetime
import time
import pandas as pd
import json
import requests
#这两个参数的默认设置都是False
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)
MP_KEY = 'e10adc3949ba59abbe56e057f20f883e' #数字签名校验码
#得到yyyy-mm-dd hh:ss:nn格式的字符串
def fmt_now_time():
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
#根据输入字符串生成一个md5串
def getmd5(str):
m1 = hashlib.md5()
m1.update(str.encode("utf-8"))
md5str = m1.hexdigest()
return md5str
def create_sig(url_in, front_type_in, time_stamp_in):
str = 'url=%s&front_type=%s&time_stamp=%s&key=%s' % (url_in, front_type_in, time_stamp_in, MP_KEY)
chenk_md5 = getmd5(str)
return chenk_md5
def Calc_Run_Time(func): #计算用时的函数装饰器
def wrapper(*args,**kwargs):
local_time = time.time(); res=func(*args, **kwargs)
print('[%s] 函数运行时间：%.3f 秒' % (func.__name__ ,time.time() - local_time)); return res
return wrapper
@Calc_Run_Time
def get_info():
request_addr = 'http://127.0.0.1:8005'
#日数据
df=pd.DataFrame()
time_stamp = str(round(time.mktime(datetime.datetime.now().timetuple())))
sig = create_sig('/info', "pc", time_stamp)
header = {}
header["Content-Type"] = 'application/json; charset=utf-8'
header["MP-SIG"] = sig
header["MP-TIMESTAMP"] = time_stamp
header["MP-FRONT-TYPE"] = "pc"
res = requests.post(url=request_addr + "/info", json={}, headers=header)
rstr = json.loads(res._content)
if 'return_result' in rstr.keys():
df = pd.DataFrame.from_dict(rstr['return_result'], orient='index')
if len(df) == 0:
return df #df为空直接返回
return df
else:
print(rstr)
def get_price(security, start_date=None, end_date=None, frequency='1d', fields=['open','close', 'low', 'high'], count=None, fq='pre'):
request_addr = 'http://127.0.0.1:8005'
time_stamp = str(round(time.mktime(datetime.datetime.now().timetuple())))
df=pd.DataFrame()
sig = create_sig('/get_price', "pc", time_stamp)
header = {}
header["Content-Type"] = 'application/json; charset=utf-8'
header["MP-SIG"] = sig
header["MP-TIMESTAMP"] = time_stamp
header["MP-FRONT-TYPE"] = "pc"
start_date=fmt_now_time() if start_date else None
end_date=fmt_now_time() if end_date else None
post_data = {'security':security, 'start_date': start_date, 'end_date': end_date, 'count': count, 'frequency': frequency,'fields':fields,'fq':fq }
try:
res = requests.post(url=request_addr + "/get_price", json=post_data, headers=header)
except:
return df
rstr = json.loads(res._content)
if 'return_result' in rstr.keys():
df = pd.DataFrame.from_dict(rstr['return_result'], orient='index')
if len(df)==0:
return df #df为空直接返回
df.index=pd.to_datetime(df.index, unit='ms');
return df
# 程序入口函数
if __name__ == "__main__":
#1分钟的数据获取
df = get_price('btc.usdt', end_date=fmt_now_time(), count=10, frequency='1m')
print(df)
#日线的数据获取
df = get_price('btc.usdt', end_date=fmt_now_time(), count=10, frequency='1d')
print(df)
#4小时的数据获取
df = get_price('btc.usdt', end_date=fmt_now_time(), count=10, frequency='4h')
print(df)
#1小时的数据获取
df = get_price('btc.usdt', end_date=fmt_now_time(), count=10, frequency='60m')
print(df)
#取到btc和eth的实时分钟线数据
df = get_info()
print(df)