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

小象实战讲义 · 量化交易Python入门：从咖啡馆到华尔街

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

💡 核心导读

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

基础读写操作

文件读取三模式

import os

# 创建示例数据文件
sample_data = """2024-01-02 09:30:00 [INFO] 系统启动
2024-01-02 09:31:15 [INFO] 用户登录成功
2024-01-02 09:32:45 [WARN] 内存使用率超过80%
2024-01-02 09:35:20 [ERROR] 数据库连接失败
2024-01-02 09:36:10 [INFO] 重连数据库成功
"""

with open('data.txt', 'w', encoding='utf-8') as f:
    f.write(sample_data)

# 块1：三种读取方式
print("=== 读取整个文件 ===")
with open('data.txt', 'r', encoding='utf-8') as f:
    content = f.read()
print(content)

print("\n=== 逐行迭代读取 ===")
with open('data.txt', 'r', encoding='utf-8') as f:
    for line in f:
        print(line.strip())

print("\n=== 读取为列表 ===")
with open('data.txt', 'r', encoding='utf-8') as f:
    lines = f.readlines()
print(f"共读取 {len(lines)} 行")
for i, line in enumerate(lines, 1):
    print(f"第{i}行: {line.strip()}")

# 清理临时文件
os.remove('data.txt')

=== 读取整个文件 ===
2024-01-02 09:30:00 [INFO] 系统启动
2024-01-02 09:31:15 [INFO] 用户登录成功
2024-01-02 09:32:45 [WARN] 内存使用率超过80%
2024-01-02 09:35:20 [ERROR] 数据库连接失败
2024-01-02 09:36:10 [INFO] 重连数据库成功


=== 逐行迭代读取 ===
2024-01-02 09:30:00 [INFO] 系统启动
2024-01-02 09:31:15 [INFO] 用户登录成功
2024-01-02 09:32:45 [WARN] 内存使用率超过80%
2024-01-02 09:35:20 [ERROR] 数据库连接失败
2024-01-02 09:36:10 [INFO] 重连数据库成功

=== 读取为列表 ===
共读取 5 行
第1行: 2024-01-02 09:30:00 [INFO] 系统启动
第2行: 2024-01-02 09:31:15 [INFO] 用户登录成功
第3行: 2024-01-02 09:32:45 [WARN] 内存使用率超过80%
第4行: 2024-01-02 09:35:20 [ERROR] 数据库连接失败
第5行: 2024-01-02 09:36:10 [INFO] 重连数据库成功

文件写入与追加

# 块2：写入和追加文件
print("=== 写入模式（覆盖） ===")
with open('output.txt', 'w', encoding='utf-8') as f:
    f.write("第一行内容\n")
    f.writelines(["第二行\n", "第三行\n"])

with open('output.txt', 'r', encoding='utf-8') as f:
    print("写入后内容:")
    print(f.read())

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

with open('output.txt', 'r', encoding='utf-8') as f:
    print("追加后内容:")
    print(f.read())

# 清理临时文件
import os
os.remove('output.txt')

=== 写入模式（覆盖） ===
写入后内容:
第一行内容
第二行
第三行


=== 追加模式 ===
追加后内容:
第一行内容
第二行
第三行
新增内容

高级处理技巧

编码检测与转换

# 块3：演示编码检测与转换（使用标准库）
import codecs

# 创建GBK编码的示例数据（在内存中）
gbk_content = "股票代码,名称,价格\n000001,平安银行,15.23\n600519,贵州茅台,1750.50"
gbk_bytes = gbk_content.encode('gbk')

# 模拟检测编码（实际环境中可能使用chardet，这里演示标准库方法）
# 假设我们已知或通过其他方式确定编码为GBK
src_encoding = 'gbk'
print(f"假设检测到的编码: {src_encoding}")

# 使用codecs解码
content = codecs.decode(gbk_bytes, src_encoding)
print(f"读取的内容:\n{content}")

# 转换为UTF-8（在内存中）
utf8_bytes = content.encode('utf-8')
print("\n转换后验证（解码UTF-8）:")
print(codecs.decode(utf8_bytes, 'utf-8'))

假设检测到的编码: gbk
读取的内容:
股票代码,名称,价格
000001,平安银行,15.23
600519,贵州茅台,1750.50

转换后验证（解码UTF-8）:
股票代码,名称,价格
000001,平安银行,15.23
600519,贵州茅台,1750.50

大文件分块处理

# 创建大文件示例数据
large_content = "交易日志数据\n" * 1000 + "结束标记"
with open('big_data.txt', 'w', encoding='utf-8') as f:
    f.write(large_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

print("分块处理大文件:")
total_size = 0
for i, chunk in enumerate(process_large_file('big_data.txt', chunk_size=500), 1):
    chunk_size = len(chunk)
    total_size += chunk_size
    print(f"第{i}块: {chunk_size}字节, 内容预览: {chunk[:50].replace(chr(10), ' ')}...")
print(f"总处理大小: {total_size}字节")

# 清理临时文件
import os
os.remove('big_data.txt')

数据结构化处理

日志文件解析示例

原始日志格式：

分块处理大文件:
第1块: 500字节, 内容预览: 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交...
第2块: 500字节, 内容预览: 志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志...
第3块: 500字节, 内容预览:  交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 ...
第4块: 500字节, 内容预览: 日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日...
第5块: 500字节, 内容预览: 据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据...
第6块: 500字节, 内容预览: 易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易...
第7块: 500字节, 内容预览: 数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数...
第8块: 500字节, 内容预览: 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交...
第9块: 500字节, 内容预览: 志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志...
第10块: 500字节, 内容预览:  交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 ...
第11块: 500字节, 内容预览: 日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日...
第12块: 500字节, 内容预览: 据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据...
第13块: 500字节, 内容预览: 易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易...
第14块: 500字节, 内容预览: 数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数据 交易日志数...
第15块: 4字节, 内容预览: 结束标记...
总处理大小: 7004字节

解析代码：

import re
from collections import defaultdict

# 创建示例日志文件
log_content = """2024-01-02 09:30:00 [INFO] 系统启动完成
2024-01-02 09:31:15 [INFO] 用户admin登录成功
2024-01-02 09:32:45 [WARN] 内存使用率85%
2024-01-02 09:35:20 [ERROR] 数据库连接超时
2024-01-02 09:36:10 [INFO] 数据库重连成功
2024-01-02 09:40:00 [DEBUG] 缓存刷新开始
2024-01-02 09:41:30 [ERROR] 交易处理失败
"""

with open('system.log', 'w', encoding='utf-8') as f:
    f.write(log_content)

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, message = match.groups()
                stats[level] += 1
                print(f"{timestamp} | {level:5s} | {message}")
    return stats

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

# 清理临时文件
import os
os.remove('system.log')

执行前后对比：

日志解析结果:
2024-01-02 09:30:00 | INFO  | 系统启动完成
2024-01-02 09:31:15 | INFO  | 用户admin登录成功
2024-01-02 09:32:45 | WARN  | 内存使用率85%
2024-01-02 09:35:20 | ERROR | 数据库连接超时
2024-01-02 09:36:10 | INFO  | 数据库重连成功
2024-01-02 09:40:00 | DEBUG | 缓存刷新开始
2024-01-02 09:41:30 | ERROR | 交易处理失败

日志级别统计:
DEBUG: 1次
ERROR: 2次
INFO: 3次
WARN: 1次

📝 动手练一练

第 1 题 · 概念应用

你的量化策略程序需要持续监控一个由外部数据源（如行情网关）生成的实时日志文件 market_feed.log，该文件会不断追加新的行情快照。程序需要实时读取并处理最新到达的日志行，同时避免因一次性读取整个大文件而占用过多内存。以下哪种文件读取方式最适合此场景？

• A. 使用 open(‘market_feed.log’, ‘r’).read() 一次性读取全部内容，处理完后再关闭文件。
• B. 使用 open(‘market_feed.log’, ‘r’).readlines() 将文件所有行读入列表，然后遍历列表。
• C. 使用 with open(‘market_feed.log’, ‘r’) as f: 打开文件后，在循环中反复调用 f.read(4096) 分块读取。
• D. 使用 with open(‘market_feed.log’, ‘r’) as f: 打开文件后，直接对文件对象 f 进行 for line in f: 迭代。

第 2 题 · 代码填空

你正在编写一个数据清洗脚本，需要读取一个来源不明的中文文本数据文件 raw_data.txt，但该文件可能使用 GBK 或 UTF-8 编码。脚本的目标是：尝试用 UTF-8 解码，如果失败则自动回退到 GBK 解码，最终把读到的内容统一按 UTF-8 重新编码并验证。请补全下面代码中的空缺部分。

import codecs

# 模拟两种不同编码的原始字节数据
utf8_bytes = "沪深300, 上证指数\n".encode('utf-8')
gbk_bytes = "净利润, 同比增长\n".encode('gbk')

def read_file_with_fallback(file_path):
    """尝试用UTF-8读取文件，失败则用GBK"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        print(f"成功以 UTF-8 读取文件")
    except UnicodeDecodeError:
        print("UTF-8 解码失败，尝试 GBK...")
        with open(file_path, 'r', encoding=______) as f:
            content = f.read()
        print(f"成功以 GBK 读取文件")
    return content

# 测试1：写入UTF-8编码的文件
with open('test_utf8.txt', 'wb') as f:
    f.write(utf8_bytes)
content1 = read_file_with_fallback('test_utf8.txt')
print(f"读取内容: {content1}")

# 测试2：写入GBK编码的文件
with open('test_gbk.txt', 'wb') as f:
    f.write(gbk_bytes)
content2 = read_file_with_fallback('test_gbk.txt')
print(f"读取内容: {content2}")

# 统一转换为UTF-8字节并验证
utf8_output = content2.encode('utf-8')
print(f"统一转换为UTF-8后的字节: {utf8_output}")
print(f"重新解码: {utf8_output.decode('utf-8')}")

# 清理临时文件
import os
os.remove('test_utf8.txt')
os.remove('test_gbk.txt')

import codecs
import os

# 模拟两种不同编码的原始字节数据
utf8_bytes = "沪深300, 上证指数\n".encode('utf-8')
gbk_bytes = "净利润, 同比增长\n".encode('gbk')

def read_file_with_fallback(file_path):
    """尝试用UTF-8读取文件，失败则用GBK"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        print(f"成功以 UTF-8 读取文件")
    except UnicodeDecodeError:
        print("UTF-8 解码失败，尝试 GBK...")
        with open(file_path, 'r', encoding='gbk') as f:
            content = f.read()
        print(f"成功以 GBK 读取文件")
    return content

# 测试1：写入UTF-8编码的文件
with open('test_utf8.txt', 'wb') as f:
    f.write(utf8_bytes)
content1 = read_file_with_fallback('test_utf8.txt')
print(f"读取内容: {content1}")

# 测试2：写入GBK编码的文件
with open('test_gbk.txt', 'wb') as f:
    f.write(gbk_bytes)
content2 = read_file_with_fallback('test_gbk.txt')
print(f"读取内容: {content2}")

# 统一转换为UTF-8字节并验证
utf8_output = content2.encode('utf-8')
print(f"统一转换为UTF-8后的字节: {utf8_output}")
print(f"重新解码: {utf8_output.decode('utf-8')}")

# 清理临时文件
os.remove('test_utf8.txt')
os.remove('test_gbk.txt')

成功以 UTF-8 读取文件
读取内容: 沪深300, 上证指数

UTF-8 解码失败，尝试 GBK...
成功以 GBK 读取文件
读取内容: 净利润, 同比增长

统一转换为UTF-8后的字节: b'\xe5\x87\x80\xe5\x88\xa9\xe6\xb6\xa6, \xe5\x90\x8c\xe6\xaf\x94\xe5\xa2\x9e\xe9\x95\xbf\n'
重新解码: 净利润, 同比增长

本章小结

📋 行动清单

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

—— 小象教研组