如何用Python实现追踪变量变化的动态数据表？

1. 是否存在解决方案？ #

当然有，完全可以实现你需要的保留历史数据、自动记录变动的动态数据表，核心思路是每次拉取新数据后，和历史数据做对比，把变动信息（拉取时间、新旧值）持久化存储起来。

2. 应搜索哪些技术方向？ #

• Pandas数据对比：学习如何用Pandas比较DataFrame的差异，提取特定字段（students、signed_up）的新旧值
• 数据持久化：
  • 轻量文件存储：CSV/Excel文件，适合小数据量场景，操作简单
  • 嵌入式数据库：SQLite，适合数据量较大、需要查询筛选的场景
• 时间戳生成：Python datetime 模块，生成每次数据拉取的时间标记

3. 代码示例 #

示例1：CSV文件存储历史记录（适合新手入门） #

这个方案操作简单，无需额外数据库，用CSV保存所有历史变动：

import pandas as pd
from datetime import datetime
import os

# 模拟从多源拉取数据的函数（替换成你的实际拉取逻辑）
def fetch_new_data():
    return pd.DataFrame({
        'students': ['Alice', 'Bob', 'Charlie'],
        'signed_up': [True, False, True]
    })

# 初始化历史文件（不存在则创建）
history_file = 'data_changes_history.csv'
if not os.path.exists(history_file):
    pd.DataFrame(columns=['pull_time', 'student', 'old_signed_up', 'new_signed_up']).to_csv(history_file, index=False)

# 读取最新全量数据（用来和新数据对比）
latest_data_file = 'latest_data.csv'
if os.path.exists(latest_data_file):
    old_data = pd.read_csv(latest_data_file)
else:
    # 第一次拉取，无旧数据，直接生成初始记录
    new_data = fetch_new_data()
    new_data.to_csv(latest_data_file, index=False)
    # 生成初始历史记录（old值设为None）
    initial_history = new_data.copy()
    initial_history['pull_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    initial_history['old_signed_up'] = None
    initial_history.rename(columns={'signed_up': 'new_signed_up'}, inplace=True)
    initial_history[['pull_time', 'students', 'old_signed_up', 'new_signed_up']].to_csv(history_file, mode='a', header=False, index=False)
    print("首次拉取数据，已生成初始历史记录")
    exit()

# 拉取新数据
new_data = fetch_new_data()

# 合并新旧数据，筛选变动记录
merged = pd.merge(old_data, new_data, on='students', how='outer', suffixes=('_old', '_new'))
changes = merged[merged['signed_up_old'] != merged['signed_up_new']]

# 记录变动到历史文件
if not changes.empty:
    changes['pull_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    changes['signed_up_old'] = changes['signed_up_old'].fillna(None)
    changes['signed_up_new'] = changes['signed_up_new'].fillna(None)
    changes[['pull_time', 'students', 'signed_up_old', 'signed_up_new']].to_csv(history_file, mode='a', header=False, index=False)
    print(f"检测到{len(changes)}条变动，已记录到历史文件")
else:
    print("本次拉取无数据变动")

# 更新最新全量数据文件
new_data.to_csv(latest_data_file, index=False)

示例2：SQLite数据库存储（适合数据量较大的场景） #

用SQLite可以更方便地查询历史记录，比如按时间范围或学生筛选：

import pandas as pd
from datetime import datetime
import sqlite3

# 模拟拉取数据函数
def fetch_new_data():
    return pd.DataFrame({
        'students': ['Alice', 'Bob', 'Charlie'],
        'signed_up': [True, False, True]
    })

# 连接SQLite数据库（不存在则自动创建）
conn = sqlite3.connect('data_history.db')

# 创建历史记录表和最新数据表（不存在则创建）
conn.execute('''
CREATE TABLE IF NOT EXISTS data_changes (
    pull_time TEXT,
    student TEXT,
    old_signed_up BOOLEAN,
    new_signed_up BOOLEAN
)
''')

conn.execute('''
CREATE TABLE IF NOT EXISTS latest_data (
    student TEXT PRIMARY KEY,
    signed_up BOOLEAN
)
''')

# 读取旧数据
old_data = pd.read_sql('SELECT * FROM latest_data', conn)

if old_data.empty:
    # 首次拉取，初始化数据库
    new_data = fetch_new_data()
    new_data.to_sql('latest_data', conn, if_exists='replace', index=False)
    # 生成初始历史记录
    initial_history = new_data.copy()
    initial_history['pull_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    initial_history['old_signed_up'] = None
    initial_history.rename(columns={'signed_up': 'new_signed_up'}, inplace=True)
    initial_history.to_sql('data_changes', conn, if_exists='append', index=False)
    print("首次拉取数据，已初始化数据库")
    conn.close()
    exit()

# 拉取新数据并对比
new_data = fetch_new_data()
merged = pd.merge(old_data, new_data, on='students', how='outer', suffixes=('_old', '_new'))
changes = merged[merged['signed_up_old'] != merged['signed_up_new']]

# 写入变动记录
if not changes.empty:
    changes['pull_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    changes['signed_up_old'] = changes['signed_up_old'].fillna(None)
    changes['signed_up_new'] = changes['signed_up_new'].fillna(None)
    changes.rename(columns={'students': 'student'}, inplace=True)
    changes[['pull_time', 'student', 'signed_up_old', 'signed_up_new']].to_sql('data_changes', conn, if_exists='append', index=False)
    print(f"检测到{len(changes)}条变动，已写入数据库")
else:
    print("本次拉取无数据变动")

# 更新最新数据
new_data.to_sql('latest_data', conn, if_exists='replace', index=False)
conn.close()

# 示例：查询所有历史记录
# conn = sqlite3.connect('data_history.db')
# history = pd.read_sql('SELECT * FROM data_changes', conn)
# print(history)
# conn.close()

内容的提问来源于stack exchange，提问作者gsgilbert