| | |
| | | import datetime |
| | | import json |
| | | |
| | | import l2_data_manager |
| | |
| | | l2_data_manager.local_today_datas[code]) > 0: |
| | | _start_index = l2_data_manager.local_today_datas[code][-1]["index"] + 1 |
| | | add_datas = l2_data_manager.L2DataUtil.get_add_data(code, datas, _start_index) |
| | | l2_data_manager.local_latest_datas[code]=add_datas |
| | | l2_data_manager.local_latest_datas[code] = add_datas |
| | | l2_data_manager.local_today_datas[code].extend(add_datas) |
| | | |
| | | path = "D:/test/2750_3.txt" |
| | |
| | | l2_data_manager.local_today_datas[code].extend(add_datas) |
| | | print(l2_data_manager.local_today_datas[code]) |
| | | |
| | | def get_space_position(self, code): |
| | | date = datetime.datetime.now().strftime("%Y-%m-%d") |
| | | path = "D:/logs/gp/l2/l2_process.{}.log".format(date) |
| | | list = [] |
| | | |
| | | with open(path, encoding="utf-8") as f: |
| | | while True: |
| | | line = f.readline() |
| | | if line: |
| | | if line.find(code) > -1: |
| | | start = line.find("处理数据范围:") |
| | | end = line.find("处理时间:") |
| | | line = (line[start:end]) |
| | | line = line[line.find(":") + 1:len(line)] |
| | | line = line.strip() |
| | | print(line) |
| | | list.append((int(line.split("-")[0]), int(line.split("-")[1]))) |
| | | else: |
| | | break |
| | | return list |
| | | |
| | | |
| | | |
| | | |
| | | if __name__ == '__main__': |
| | | L2DataTest().test_concat_l2_data() |
| | | L2DataTest().get_space_position("002094") |