from utils import tool
|
|
|
def __get_transaction_datas():
|
transaction_dict = {}
|
with open("D:\\文件传输\\交易\\日志文件\\transaction.2024-01-12.log", mode="r") as f:
|
lines = f.readlines()
|
for line in lines:
|
si = line.find("#")
|
code = line[si - 6:si]
|
value = line[si + 1:]
|
if code not in transaction_dict:
|
transaction_dict[code] = []
|
transaction_dict[code].extend(eval(value))
|
return transaction_dict
|
|
|
def __convert_time(time_str):
|
time_str = str(time_str)
|
if time_str.startswith("9"):
|
time_str = f"0{time_str}"
|
ms = "{:0<3}".format(time_str[6:])
|
time_ = f"{time_str[0:2]}:{time_str[2:4]}:{time_str[4:6]}"
|
return f"{time_}", int(ms)
|
|
import matplotlib.pyplot as plt
|
def show_data(datas):
|
|
|
# 示例数据
|
x = [t[0] for t in datas] # x轴数据
|
y = [t[1] for t in datas] # x轴数据
|
|
# 创建折线图
|
plt.plot(x, y)
|
|
# 添加标题和轴标签
|
plt.title("折线图示例")
|
plt.xlabel("X轴")
|
plt.ylabel("Y轴")
|
|
# 显示图形
|
plt.show()
|
|
|
if __name__ == "__main__":
|
transaction_datas_dict = __get_transaction_datas()
|
for k in transaction_datas_dict:
|
for d in transaction_datas_dict[k]:
|
if d[2] > 10000:
|
print(d)
|
|
if __name__ == "__main__1":
|
code = "603663"
|
transaction_datas_dict = __get_transaction_datas()
|
datas = transaction_datas_dict.get(code)
|
# 100ms计算一次速率
|
start_time_ms = None
|
total_num = 0
|
show_datas=[]
|
for d in datas:
|
# 转为ms
|
time_str, ms = __convert_time(d[3])
|
time_as_ms = tool.get_time_as_second(time_str) * 1000 + ms
|
if start_time_ms is None:
|
start_time_ms = time_as_ms
|
|
total_num += d[2]
|
if time_as_ms - start_time_ms >= 100:
|
ms = start_time_ms % 1000
|
time_str = tool.time_seconds_format(start_time_ms // 1000)
|
show_datas.append((f"{time_str}.{ms}",round(int(total_num / (time_as_ms - start_time_ms) * 100) * d[1]/10000, 2)))
|
# print(f"{time_str}.{ms}", f"{round(int(total_num / (time_as_ms - start_time_ms) * 100) * d[1]/10000, 2)}w")
|
# 结束计算
|
total_num = 0
|
start_time_ms = None
|
show_data(show_datas)
|
