| | |
|---|
| | | import os |
|---|
| | | import sys |
|---|
| | | import time |
|---|
| | | from multiprocessing import Pool |
|---|
| | | |
|---|
| | | import pandas as pd |
|---|
| | | |
|---|
| | | from db import mysql_data_delegate as mysql_data |
|---|
| | |
|---|
| | | # 将文件写入到文本 |
|---|
| | | writer.writerow(row) |
|---|
| | | |
|---|
| | | def test1(args): |
|---|
| | | index, df = args |
|---|
| | | print(index) |
|---|
| | | |
|---|
| | | def pre_process_transactions(csv_path="E:/测试数据/Transaction_Test.csv"): |
|---|
| | | def str_to_float(s): |
|---|
| | |
|---|
| | | 'EndPrice': group['TradePrice'].iloc[-1] |
|---|
| | | }) |
|---|
| | | |
|---|
| | | |
|---|
| | | |
|---|
| | | |
|---|
| | | dtype = { |
|---|
| | | 'SecurityID': 'category', # 低基数分类数据 |
|---|
| | | } |
|---|
| | | chunk_size = 100000 |
|---|
| | | chunk_size = 10000 |
|---|
| | | # 创建DataFrame |
|---|
| | | chunks = pd.read_csv(csv_path, chunksize=chunk_size) |
|---|
| | | indexed_data = list(enumerate(chunks)) |
|---|
| | | # 新写法 |
|---|
| | | with Pool(processes=4) as pool: |
|---|
| | | pool.map(test1, indexed_data) |
|---|
| | | |
|---|
| | | result_list = [] |
|---|
| | | index = 0 |
|---|
| | | for df in chunks: |
|---|
| | | index += 1 |
|---|
| | | for chunk_index, chunk in enumerate(chunks): |
|---|
| | | df = chunk.copy() |
|---|
| | | index = chunk_index + 1 |
|---|
| | | child_path = csv_path.replace(".csv", f"_{index}.csv") |
|---|
| | | if os.path.exists(child_path): |
|---|
| | | continue |
|---|
| | |
|---|
| | | df = df[df["SecurityID"].str.startswith(("30", "00", "60"), na=False)] |
|---|
| | | # 计算成交金额 |
|---|
| | | df['TradeAmount'] = df['TradePrice'] * df['TradeVolume'] |
|---|
| | | |
|---|
| | | |
|---|
| | | # 按SecurityID和BuyNo分组 |
|---|
| | | grouped = df.groupby(['SecurityID', 'BuyNo']) |
|---|
| | |
|---|
| | | |
|---|
| | | # [ExchangeID,SecurityID,MainSeq,SubSeq,TickTime,TickType,BuyNo,SellNo,Price,Volume,TradeMoney,Side,TradeBSFlag,MDSecurityStat,Info1,Info2,Info3,LocalTimeStamp] |
|---|
| | | |
|---|
| | | chunk_size = 200000 |
|---|
| | | chunk_size = 10000 |
|---|
| | | # 创建DataFrame |
|---|
| | | chunks = pd.read_csv(csv_path, chunksize=chunk_size) |
|---|
| | | result_list = [] |
|---|
| | |
|---|
| | | print(f"处理完毕，总共{index}批") |
|---|
| | | |
|---|
| | | |
|---|
| | | if __name__ == '__main__1': |
|---|
| | | if __name__ == '__main__': |
|---|
| | | # df = pd.read_csv(f"E:/测试数据/Transaction_Test.csv") |
|---|
| | | pre_process_ngtstick() |
|---|
| | | pre_process_transactions() |
|---|
| | | |
|---|
| | | # 命令模式 /home/userzjj/app/gp-server/l2_data_parser Transaction 2025-05-08 |
|---|
| | | # 解析大单： /home/userzjj/app/gp-server/l2_data_parser ExtractDealBigOrder 2025-05-09 /home/userzjj/最终成交数据20250509.txt 000555 |
|---|
| | | if __name__ == '__main__': |
|---|
| | | if __name__ == '__main__1': |
|---|
| | | if len(sys.argv) > 1: |
|---|
| | | params = sys.argv[1:] |
|---|
| | | print("接收的参数", params) |
|---|
