股票日期面板处理

任务书与输出要求

你得到一张故意包含乱序、重复股票日期、缺失日期和缺失价格的行情表，需要交付一份能够复核的特征数据。任务不是估计策略赚钱多少，而是证明价格如何清洗、窗口怎样定义和特征在何时可知。输出重复键审计表、清洗后的特征表、股票覆盖表、行业观测收益汇总和价格及滞后均线图。

每行最终必须对应唯一的股票和日期，行业只是属性，不能成为放大行数的另一张事实表。重复记录按本课教学规则保留输入最后一条，原始数据和全部冲突行都保留。这个规则不代表价格最后发布版本，真实项目需明确更新时间或来源优先级。缺失价格保留为空，图中自然断开，不能把未知改为零后计算巨大虚假收益。

起始代码与数据检查

先读取原表并标记重复键，把冲突行输出，再在副本去重。解析日期，按股票日期稳定排序，验证键唯一、日期有效、非空价格正数。这几步先做完才进入收益与窗口计算。覆盖表每股票给出观测行数、缺失收盘数、首末日期，但不把日历跨度自动解释为交易日数量。

起始代码：

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
raw=load("panel")
keys=["symbol","date"]
audit=raw.loc[raw.duplicated(keys,keep=False)].copy()
# TODO: 保留审计，创建清洗副本；解析日期并排序。
# TODO: 在股票组内计算前值、间隔天数、窗口和滞后信号。

观察期内缺少某日可能是周末、停牌或数据缺口；本项目没有完整交易日历，因此只报告观测间隔天数，并将收益命名为 observed_return，避免暗示每条都是单日收益。

特征和时点逐步演算

假设同一股票观测收盘依次十、十二、十五，第三条三观测均线为三十七除以三。若第三条开盘时决定交易，这个均线包含第三条收盘，尚不可知；第四条观测开始前，才能使用它。因此先算均线，再在股票组内 shift 一行，作为 signal_before_observation。每组第一条前值和初始不足窗口的信号都应为空。

缺失 close 会让对应窗口有效观测不足，min_periods=3 保持缺失。收益率只使用相邻记录的原始前值，不自动跳过缺失价格；这样把缺口暴露出来。生产策略还应处理公告发布时间、复权信息和执行时间，本项目只验证基本时点机制，不把这个特征直接当成交易建议。

报告、SQL 对照与图形

行业汇总按日期和行业计算观测收益均值，并同时输出有效收益数。不同股票观测间隔可能不同，所以这个均值只能描述目前记录的区间收益，不能宣称行业单日可交易收益。若需要单日比较，应提供共同交易日历、补齐股票日期和明确缺失规则后重新计算。

SQL 实现用 LAG 和窗口聚合时，必须 PARTITION BY symbol，ORDER BY date，并在唯一键清洗后执行。ROWS 最近三行与按日历时间的窗口不是同一个概念。本项目图形用两幅共享日期坐标的图分别展示原始收盘和滞后均线，标记观测点，保留价格缺失造成的断线，标题和轴标签说明实际含义。

验收标准与边界

完整答案为本章脚本。验收检查股票日期唯一、清洗行数不超过输入、审计保存全部重复记录、首行前值为空、信号不使用当前收盘；所有文件仅写到 outputs。图片必须实际打开查看，确认图例、日期和轴标签可读，不重叠遮挡，输出 CSV 可以再次读取。

分别检查只有一条记录的股票、所有价格缺失、日期间隔很大和重复价格冲突。若一组不足三条，均线与信号为空是预期结果，不能为凑出信号降低最小观测要求而不说明。说明书最后列出缺失交易日历、未考虑复权和来源版本等限制，使下一位分析师知道哪些结论能够使用、哪些需要补充数据。

版本与复现约定

本课程 pandas 示例以 2.3.3 为验证基线，页面运行记录给出实际版本。pandas 3.x 默认使用 Copy-on-Write：从表中取出的对象发生修改时，不据此假定原表同步变化。跨版本都应使用一次完整的 loc 赋值表达式；需要独立工作表时显式 copy。阅读网上的旧示例时，先核对版本、输入类型和索引，再判断结果能否复现。

交互演示：观察机制

完整示例与实际输出

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
raw=load("panel").copy()
audit=raw.loc[raw.duplicated(["symbol","date"],keep=False)].copy()
audit.to_csv(OUT/"project_panel_duplicates.csv",index=False)
d=raw.drop_duplicates(["symbol","date"],keep="last").copy()
d["date"]=pd.to_datetime(d["date"],errors="raise")
d=d.sort_values(["symbol","date"]).reset_index(drop=True)
assert not d.duplicated(["symbol","date"]).any()
assert d.loc[d["close"].notna(),"close"].gt(0).all()
g=d.groupby("symbol",sort=False)
d["previous_close"]=g["close"].shift(1)
d["gap_days"]=g["date"].diff().dt.days
d["observed_return"]=d["close"]/d["previous_close"]-1
d["mean_3"]=g["close"].transform(lambda s:s.rolling(3,min_periods=3).mean())
d["signal_before_observation"]=d.groupby("symbol")["mean_3"].shift(1)
d["missing_close"]=d["close"].isna()
assert g.head(1)["previous_close"].isna().all()
assert d.groupby("symbol").head(1)["signal_before_observation"].isna().all()
coverage=d.groupby("symbol").agg(rows=("date","size"),missing_close=("missing_close","sum"),first_date=("date","min"),last_date=("date","max"))
industry=d.groupby(["date","industry"])["observed_return"].agg(mean_return="mean",valid_returns="count").reset_index()
d.to_csv(OUT/"project_panel_features.csv",index=False)
coverage.to_csv(OUT/"project_panel_coverage.csv")
industry.to_csv(OUT/"project_panel_industry.csv",index=False)
fig,axes=plt.subplots(2,1,figsize=(8,6),sharex=True)
for symbol,part in d.groupby("symbol"):
    axes[0].plot(part["date"],part["close"],marker="o",label=symbol)
    axes[1].plot(part["date"],part["signal_before_observation"],label=symbol)
axes[0].set_ylabel("Observed close")
axes[1].set_ylabel("Lagged 3-observation mean")
axes[1].set_xlabel("Date")
axes[0].legend();axes[1].legend()
fig.autofmt_xdate();fig.tight_layout()
fig.savefig(OUT/"project_panel.png",dpi=160)
plt.close(fig)
print("原始行 / 清洗行 / 重复审计行:",len(raw),len(d),len(audit))
print(coverage.to_string())
print(d[["symbol","date","gap_days","observed_return","signal_before_observation"]].head(10).to_string(index=False))

实际运行通过CPU Python execution.

原始行 / 清洗行 / 重复审计行: 144 143 2
        rows  missing_close first_date  last_date
symbol                                           
AAA       36              0 2026-01-01 2026-02-19
BBB       35              0 2026-01-01 2026-02-19
CCC       36              1 2026-01-01 2026-02-19
DDD       36              0 2026-01-01 2026-02-19
symbol       date  gap_days  observed_return  signal_before_observation
   AAA 2026-01-01       NaN              NaN                        NaN
   AAA 2026-01-02       1.0         0.028904                        NaN
   AAA 2026-01-05       3.0         0.017177                        NaN
   AAA 2026-01-06       1.0         0.011987                  27.047533
   AAA 2026-01-07       1.0         0.003178                  27.567500
   AAA 2026-01-08       1.0        -0.004930                  27.862867
   AAA 2026-01-09       1.0         0.020255                  27.956733
   AAA 2026-01-12       3.0        -0.001395                  28.128600
   AAA 2026-01-13       1.0        -0.008628                  28.257633
   AAA 2026-01-14       1.0         0.000575                  28.351000

五道自测：先回答，再核对

两项动手任务

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
raw=load("panel").copy()
audit=raw.loc[raw.duplicated(["symbol","date"],keep=False)].copy()
audit.to_csv(OUT/"project_panel_duplicates.csv",index=False)
d=raw.drop_duplicates(["symbol","date"],keep="last").copy()
d["date"]=pd.to_datetime(d["date"],errors="raise")
d=d.sort_values(["symbol","date"]).reset_index(drop=True)
assert not d.duplicated(["symbol","date"]).any()
assert d.loc[d["close"].notna(),"close"].gt(0).all()
g=d.groupby("symbol",sort=False)
d["previous_close"]=g["close"].shift(1)
d["gap_days"]=g["date"].diff().dt.days
d["observed_return"]=d["close"]/d["previous_close"]-1
d["mean_3"]=g["close"].transform(lambda s:s.rolling(3,min_periods=3).mean())
d["signal_before_observation"]=d.groupby("symbol")["mean_3"].shift(1)
d["missing_close"]=d["close"].isna()
assert g.head(1)["previous_close"].isna().all()
assert d.groupby("symbol").head(1)["signal_before_observation"].isna().all()
coverage=d.groupby("symbol").agg(rows=("date","size"),missing_close=("missing_close","sum"),first_date=("date","min"),last_date=("date","max"))
industry=d.groupby(["date","industry"])["observed_return"].agg(mean_return="mean",valid_returns="count").reset_index()
d.to_csv(OUT/"project_panel_features.csv",index=False)
coverage.to_csv(OUT/"project_panel_coverage.csv")
industry.to_csv(OUT/"project_panel_industry.csv",index=False)
fig,axes=plt.subplots(2,1,figsize=(8,6),sharex=True)
for symbol,part in d.groupby("symbol"):
    axes[0].plot(part["date"],part["close"],marker="o",label=symbol)
    axes[1].plot(part["date"],part["signal_before_observation"],label=symbol)
axes[0].set_ylabel("Observed close")
axes[1].set_ylabel("Lagged 3-observation mean")
axes[1].set_xlabel("Date")
axes[0].legend();axes[1].legend()
fig.autofmt_xdate();fig.tight_layout()
fig.savefig(OUT/"project_panel.png",dpi=160)
plt.close(fig)
print("原始行 / 清洗行 / 重复审计行:",len(raw),len(d),len(audit))
print(coverage.to_string())
print(d[["symbol","date","gap_days","observed_return","signal_before_observation"]].head(10).to_string(index=False))

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
s=pd.Series([10.,12.,15.,18.])
signal=s.rolling(3,min_periods=3).mean().shift(1)
assert signal.iloc[:3].isna().all()
assert signal.iloc[3]==37/3
print(signal.to_list())

官方资料与掌握状态

• pandas 窗口
• Matplotlib 子图

完成两项任务后，按独立完成程度记录。状态会尽量保存在当前浏览器，可导出备份。