第3周跟练：分组、合并与SQL窗口函数

def summarize(frame):
    ordered = validate_prices(frame)  # first/last取决于顺序，先排序再分组
    grouped = ordered.groupby("symbol", sort=True)
    # agg把每个标的压缩成一行。size数行，count只数该列非缺失值。
    summary = grouped.agg(
        rows=("close", "size"),
        first_close=("close", "first"),
        last_close=("close", "last"),
    ).reset_index()
    summary["period_return"] = summary["last_close"] / summary["first_close"] - 1
    # transform把组统计广播回原来的每一行，所以仍有10行。
    detail = ordered.copy()
    detail["rows_in_symbol"] = grouped["close"].transform("size")
    # 日期分组回答“这一天有几条记录”，与按标的统计是不同的问题。
    daily = ordered.groupby("date", as_index=False).agg(rows=("symbol", "size"))
    return summary, detail, daily

from week3 import read_prices, summarize, save

prices = read_prices()
summary, detail, daily = summarize(prices)
print(summary.to_string(index=False))
print(detail.to_string(index=False))
print(daily.to_string(index=False))
save(summary, "summary.csv")
save(detail, "group-detail.csv")
save(daily, "daily-count.csv")

def join_sectors(frame, sectors):
    ordered = validate_prices(frame)
    if list(sectors.columns) != ["symbol", "sector"]:
        raise ValueError("行业表需要symbol,sector两列")
    right = sectors.copy()
    for column in ["symbol", "sector"]:
        right[column] = right[column].astype("string").str.strip()
        if right[column].isna().any() or right[column].eq("").any():
            raise ValueError(f"行业表{column}不能缺失；没有映射时保留左表待查")
    # 多条价格 -> 一条行业映射。右表重复时主动报MergeError，不静默放大样本。
    result = ordered.merge(
        right, how="left", on="symbol", validate="many_to_one", indicator=True
    )
    if len(result) != len(ordered) or result.duplicated(KEYS).any():
        raise ValueError("合并改变了行数或唯一键")
    # left_only不是删除理由。保留记录，并让学习者看见未匹配状态。
    return result

import pandas as pd
from week3 import ROOT, read_prices, join_sectors, save

prices = read_prices()
sectors = pd.read_csv(ROOT / "sectors.csv", dtype="string", encoding="utf-8-sig")
joined = join_sectors(prices, sectors)
print(joined.to_string(index=False))
print("匹配状态：", joined["_merge"].value_counts().to_dict())
save(joined, "joined.csv")

# 缺失行业也作为一组显示，避免groupby默认丢掉缺失键让记录消失。
coverage = joined.groupby(["date", "sector"], dropna=False, as_index=False).agg(
    rows=("symbol", "size")
)
print(coverage.to_string(index=False))
save(coverage, "sector-coverage.csv")

# 故意把B的行业复制一次：这里捕获预期错误，所以整个演示仍正常退出。
bad = pd.concat([sectors, sectors.iloc[[-1]]], ignore_index=True)
try:
    join_sectors(prices, bad)
except pd.errors.MergeError:
    print("预期拒绝：右表symbol重复，many_to_one检查生效")
else:
    raise AssertionError("重复右键竟然未被拒绝")

# 演示未匹配：只保留A映射，B的5条价格应继续保留，sector缺失。
missing = join_sectors(prices, sectors[sectors["symbol"].eq("A")])
print("只保留A映射后的行数/未匹配数：", len(missing), missing["_merge"].eq("left_only").sum())

def pandas_returns(frame):
    result = validate_prices(frame)
    result["return"] = result.groupby("symbol")["close"].pct_change(fill_method=None)
    return result[KEYS + ["return"]]


def sql_returns(frame, query=None):
    ordered = validate_prices(frame)
    if sqlite3.sqlite_version_info < (3, 25, 0):
        raise RuntimeError("需要SQLite 3.25+以支持窗口函数；请更新Python")
    if query is None:
        query = (ROOT / "returns.sql").read_text(encoding="utf-8")
    with closing(sqlite3.connect(":memory:")) as connection:
        # closing确保退出时关闭连接，丢弃内存库；无需用户名或数据库服务。
        connection.execute("""CREATE TABLE prices (
            date TEXT NOT NULL, symbol TEXT NOT NULL,
            close REAL NOT NULL CHECK(close > 0),
            PRIMARY KEY(date, symbol)
        )""")
        # ?是参数占位符；数据通过参数传入，不拼接到SQL语句中。
        connection.executemany(
            "INSERT INTO prices(date,symbol,close) VALUES (?,?,?)",
            ordered[KEYS + ["close"]].itertuples(index=False, name=None),
        )
        return pd.read_sql_query(query, connection)

import sqlite3
from week3 import read_prices, sql_returns, save

print("SQLite版本：", sqlite3.sqlite_version)
prices = read_prices()
# 先运行普通聚合：按标的归并成2行，COUNT(*)包括所有行。
counts = sql_returns(prices, """
    SELECT symbol, COUNT(*) AS rows, COUNT(close) AS non_missing_prices
    FROM prices GROUP BY symbol ORDER BY symbol
""")
print("GROUP BY聚合：\n", counts.to_string(index=False))
# 再读取并运行returns.sql。只修改SQL文件即可练习窗口，不用重写连接代码。
returns = sql_returns(prices)
print("窗口函数保留逐条记录：\n", returns.to_string(index=False))
save(returns, "sql-returns.csv")

def compare_returns(left, right):
    """先比键，再比缺失位置，最后比数值；不能用inner join或fillna(0)掩盖错误。"""
    for frame in [left, right]:
        if not set(KEYS + ["return"]).issubset(frame.columns) or frame.empty:
            raise ValueError("核对表缺少必要列或没有记录")
        if frame[KEYS].isna().any().any() or frame.duplicated(KEYS).any():
            raise ValueError("核对表有缺失键或重复键")
    result = left[KEYS + ["return"]].merge(
        right[KEYS + ["return"]], on=KEYS, how="outer",
        validate="one_to_one", indicator=True, suffixes=("_pandas", "_sql"),
    )
    if not result["_merge"].eq("both").all():
        raise ValueError("两边的键不一致：存在丢行或多行")
    a, b = result["return_pandas"], result["return_sql"]
    if not a.isna().equals(b.isna()):
        raise ValueError("两边缺失位置不一致，不能把首条缺失填0")
    present = a.notna()
    # 浮点数存在微小舍入差异，使用明确容差；无限值始终拒绝。
    passed = [math.isfinite(x) and math.isfinite(y) and
              math.isclose(x, y, rel_tol=1e-9, abs_tol=1e-12)
              for x, y in zip(a[present], b[present])]
    if not all(passed):
        raise ValueError("收益数值不一致或含无限值")
    result["abs_diff"] = (a - b).abs()  # 两边都缺失时差值也保持缺失
    result["passed"] = True
    return result.sort_values(["symbol", "date"]).reset_index(drop=True)

from week3 import read_prices, pandas_returns, sql_returns, compare_returns, save

prices = read_prices()
comparison = compare_returns(pandas_returns(prices), sql_returns(prices))
print(comparison.to_string(index=False))
print("逐键核对通过：", len(comparison), "行")
print("两边各自缺失收益：", comparison["return_pandas"].isna().sum())
save(comparison, "compare.csv")
# 注意：两种实现一致，不证明数据来源正确，也不证明中间没有漏日期。

"""第3周共用函数。输入文件只读；结果写入outputs。数据全部是虚构教学数据。"""
from pathlib import Path
from contextlib import closing
import math
import sqlite3  # Python自带，无需pip install sqlite3，也无需安装数据库服务器
import pandas as pd

ROOT = Path(__file__).resolve().parent
KEYS = ["date", "symbol"]


# region read
def validate_prices(frame):
    """约定：每个日期+标的一条记录；价格完整、有限且为正。返回排序后的副本。"""
    if list(frame.columns) != ["date", "symbol", "close"] or frame.empty:
        raise ValueError("价格表需要非空的date,symbol,close三列，且顺序一致")
    work = frame.copy()
    for column in KEYS:
        work[column] = work[column].astype("string").str.strip()
        if work[column].isna().any() or work[column].eq("").any():
            raise ValueError(f"{column}键不能缺失")
    # ISO格式使SQLite文字排序等同本教程的日期排序；不处理时区或时间戳。
    if not work["date"].str.fullmatch(r"\d{4}-\d{2}-\d{2}").all():
        raise ValueError("日期必须为YYYY-MM-DD")
    dates = pd.to_datetime(work["date"], format="%Y-%m-%d", errors="raise")
    if dates.isna().any():
        raise ValueError("日期不能为NaT")
    work["date"] = dates.dt.strftime("%Y-%m-%d")
    work["close"] = pd.to_numeric(work["close"], errors="raise").astype(float)
    if not work["close"].map(lambda v: math.isfinite(v) and v > 0).all():
        raise ValueError("价格必须完整、有限且大于0；回第2周核验，不能填0")
    if work.duplicated(KEYS).any():
        raise ValueError("date+symbol重复，不能决定上一条是哪条")
    return work.sort_values(["symbol", "date"]).reset_index(drop=True)


def read_prices(name="cleaned.csv"):
    filename = ROOT / name
    if not filename.exists():
        raise FileNotFoundError(f"找不到{filename}；请复制第2周outputs/cleaned.csv到本目录")
    frame = pd.read_csv(filename, dtype="string", encoding="utf-8-sig", keep_default_na=False)
    if not isinstance(frame.index, pd.RangeIndex):
        raise ValueError("CSV列数异常，检查逗号和表头")
    return validate_prices(frame)
# endregion read


# region group
def summarize(frame):
    ordered = validate_prices(frame)  # first/last取决于顺序，先排序再分组
    grouped = ordered.groupby("symbol", sort=True)
    # agg把每个标的压缩成一行。size数行，count只数该列非缺失值。
    summary = grouped.agg(
        rows=("close", "size"),
        first_close=("close", "first"),
        last_close=("close", "last"),
    ).reset_index()
    summary["period_return"] = summary["last_close"] / summary["first_close"] - 1
    # transform把组统计广播回原来的每一行，所以仍有10行。
    detail = ordered.copy()
    detail["rows_in_symbol"] = grouped["close"].transform("size")
    # 日期分组回答“这一天有几条记录”，与按标的统计是不同的问题。
    daily = ordered.groupby("date", as_index=False).agg(rows=("symbol", "size"))
    return summary, detail, daily
# endregion group


# region merge
def join_sectors(frame, sectors):
    ordered = validate_prices(frame)
    if list(sectors.columns) != ["symbol", "sector"]:
        raise ValueError("行业表需要symbol,sector两列")
    right = sectors.copy()
    for column in ["symbol", "sector"]:
        right[column] = right[column].astype("string").str.strip()
        if right[column].isna().any() or right[column].eq("").any():
            raise ValueError(f"行业表{column}不能缺失；没有映射时保留左表待查")
    # 多条价格 -> 一条行业映射。右表重复时主动报MergeError，不静默放大样本。
    result = ordered.merge(
        right, how="left", on="symbol", validate="many_to_one", indicator=True
    )
    if len(result) != len(ordered) or result.duplicated(KEYS).any():
        raise ValueError("合并改变了行数或唯一键")
    # left_only不是删除理由。保留记录，并让学习者看见未匹配状态。
    return result
# endregion merge


# region sql
def pandas_returns(frame):
    result = validate_prices(frame)
    result["return"] = result.groupby("symbol")["close"].pct_change(fill_method=None)
    return result[KEYS + ["return"]]


def sql_returns(frame, query=None):
    ordered = validate_prices(frame)
    if sqlite3.sqlite_version_info < (3, 25, 0):
        raise RuntimeError("需要SQLite 3.25+以支持窗口函数；请更新Python")
    if query is None:
        query = (ROOT / "returns.sql").read_text(encoding="utf-8")
    with closing(sqlite3.connect(":memory:")) as connection:
        # closing确保退出时关闭连接，丢弃内存库；无需用户名或数据库服务。
        connection.execute("""CREATE TABLE prices (
            date TEXT NOT NULL, symbol TEXT NOT NULL,
            close REAL NOT NULL CHECK(close > 0),
            PRIMARY KEY(date, symbol)
        )""")
        # ?是参数占位符；数据通过参数传入，不拼接到SQL语句中。
        connection.executemany(
            "INSERT INTO prices(date,symbol,close) VALUES (?,?,?)",
            ordered[KEYS + ["close"]].itertuples(index=False, name=None),
        )
        return pd.read_sql_query(query, connection)
# endregion sql


# region compare
def compare_returns(left, right):
    """先比键，再比缺失位置，最后比数值；不能用inner join或fillna(0)掩盖错误。"""
    for frame in [left, right]:
        if not set(KEYS + ["return"]).issubset(frame.columns) or frame.empty:
            raise ValueError("核对表缺少必要列或没有记录")
        if frame[KEYS].isna().any().any() or frame.duplicated(KEYS).any():
            raise ValueError("核对表有缺失键或重复键")
    result = left[KEYS + ["return"]].merge(
        right[KEYS + ["return"]], on=KEYS, how="outer",
        validate="one_to_one", indicator=True, suffixes=("_pandas", "_sql"),
    )
    if not result["_merge"].eq("both").all():
        raise ValueError("两边的键不一致：存在丢行或多行")
    a, b = result["return_pandas"], result["return_sql"]
    if not a.isna().equals(b.isna()):
        raise ValueError("两边缺失位置不一致，不能把首条缺失填0")
    present = a.notna()
    # 浮点数存在微小舍入差异，使用明确容差；无限值始终拒绝。
    passed = [math.isfinite(x) and math.isfinite(y) and
              math.isclose(x, y, rel_tol=1e-9, abs_tol=1e-12)
              for x, y in zip(a[present], b[present])]
    if not all(passed):
        raise ValueError("收益数值不一致或含无限值")
    result["abs_diff"] = (a - b).abs()  # 两边都缺失时差值也保持缺失
    result["passed"] = True
    return result.sort_values(["symbol", "date"]).reset_index(drop=True)
# endregion compare


def save(frame, name):
    folder = ROOT / "outputs"
    folder.mkdir(exist_ok=True)
    frame.to_csv(folder / name, index=False, encoding="utf-8-sig")
    print(f"已写入 {folder / name}（{len(frame)}行）")