TXT数据流操作：处理金融交易日志

TXT数据流操作：处理金融交易日志

小象量化极客小册 · 华尔街量化分析先锋系列

在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

小象量化极客小册 · 华尔街量化分析先锋系列 在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

💡 核心导读

TXT文件是最基础的文本存储格式，掌握其操作是数据采集与处理的必备技能。本章将系统讲解Python处理TXT文件的5大核心操作：文件读取、写入、追加、行级处理以及编码转换。通过实战演练，您将学会使用open()函数的不同模式、with语句管理资源、处理多编码文本等实用技巧，并掌握大文件的分块读取策略。

基础读写操作

文件读取三模式

with open('data.txt', 'r', encoding='utf-8') as f:
    content = f.read()  # 返回整个文件内容的字符串

with open('data.txt', 'r', encoding='utf-8') as f:
    for line in f:      # 迭代器方式逐行处理
        print(line.strip())

with open('data.txt', 'r', encoding='utf-8') as f:
    lines = f.readlines()  # 返回包含换行符的列表

文件写入与追加

with open('output.txt', 'w', encoding='utf-8') as f:
    f.write("第一行内容\n")
    f.writelines(["第二行\n", "第三行\n"])

with open('output.txt', 'a', encoding='utf-8') as f:
    f.write("新增内容\n")

💡 概念小测：文件写模式

量化策略引擎在写入每日成交记录时，应该使用以下哪种文件打开模式以防止覆盖之前历史日的成交日志？

• ‘w’（写入模式），因为这能保证文件被清除并从头开始写。
• ‘a’（追加模式），在文件末尾追加最新的成交数据，保留历史。
• ‘r+‘（读写模式），这会自动定位到文件开头，覆盖第一行。
• ‘x’（独占写入），这会清空系统内存中的缓存，加快写入。

📝 选择题

💡 概念小测：文件写模式 量化策略引擎在写入每日成交记录时，应该使用以下哪种文件打开模式以防止覆盖之前历史日的成交日志？

• A. ‘w’（写入模式），因为这能保证文件被清除并从头开始写。
• B. ‘a’（追加模式），在文件末尾追加最新的成交数据，保留历史。
• C. ‘r+‘（读写模式），这会自动定位到文件开头，覆盖第一行。
• D. ‘x’（独占写入），这会清空系统内存中的缓存，加快写入。

答案：B（‘a’（追加模式），在文件末尾追加最新的成交数据，保留历史。）

高级处理技巧

编码检测与转换

import chardet

def detect_encoding(file_path):
    with open(file_path, 'rb') as f:
        rawdata = f.read(1024)  # 读取前1KB用于检测
    return chardet.detect(rawdata)['encoding']

src_encoding = detect_encoding('gbk_file.txt')
with open('gbk_file.txt', 'r', encoding=src_encoding) as f_in:
    content = f_in.read()
with open('utf8_file.txt', 'w', encoding='utf-8') as f_out:
    f_out.write(content)

大文件分块处理

def process_large_file(file_path, chunk_size=1024):
    """处理大文件的生成器函数"""
    with open(file_path, 'r', encoding='utf-8') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

for i, chunk in enumerate(process_large_file('big_data.txt')):
    print(f"处理第{i+1}个数据块，大小:{len(chunk)}字节")

数据结构化处理

日志文件解析示例

原始日志格式：

2023-08-20 14:30:45 [INFO] 用户登录 user_id=1001
2023-08-20 14:31:02 [ERROR] 连接超时 code=408

解析代码：

import re
from collections import defaultdict

log_pattern = re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \[(\w+)\] (.+)')

def parse_logs(file_path):
    stats = defaultdict(int)
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            match = log_pattern.match(line.strip())
            if match:
                timestamp, level, messpe_ratio = match.groups()
                stats[level] += 1
                print(f"{timestamp} | {level:5s} | {messpe_ratio}")
    return stats

log_stats = parse_logs('system.log')
print("\n日志统计结果:")
for level, count in log_stats.items():
    print(f"{level}: {count}次")

执行前后对比：

原始数据：
2023-08-20 14:30:45 [INFO] 用户登录 user_id=1001
2023-08-20 14:31:02 [ERROR] 连接超时 code=408

解析后输出：
2023-08-20 14:30:45 | INFO  | 用户登录 user_id=1001
2023-08-20 14:31:02 | ERROR | 连接超时 code=408

日志统计结果:
INFO: 1次
ERROR: 1次

💡 代码填空：逐行追加

补全代码，将一条最新的股票成交记录追加写入到本地的交易文本中：

new_trade = "2026-06-09,AAPL,Buy,100,150.5\n"
with open('trades.txt', '{blank}', encoding='utf-8') as f:
    f.write(new_trade)

• w
• a
• r

✏️ 填空题

💡 代码填空：逐行追加 补全代码，将一条最新的股票成交记录追加写入到本地的交易文本中：

new_trade = "2026-06-09,AAPL,Buy,100,150.5\n"
# 以追加模式打开文件
with open('trades.txt', '___', encoding='utf-8') as f:
    f.write(new_trade)

答案：a

本章小结

行动清单

• 创建一个包含多行文本的TXT文件，分别用三种读取模式提取内容并比较差异
• 编写一个日志分析脚本，统计不同日志级别（INFO/WARN/ERROR）的出现频率
• 下载一个GBK编码的文本文件，使用编码检测自动转换为UTF-8格式

—— 小象量化教研组