在资产定价数据中保留匿名公司身份
代码 《交易机器学习》
总结
该数据工具支持使用Chen、Pelger和Zhu资产定价数据集，其中包含股票收益、公司特征、宏观经济序列和拆分前的特征张量。其转换方法根据已发布的训练、验证和测试张量生成Parquet文件。由于存档数据在每个区块中提供了持续匿名的公司位置，却没有跨区块映射，转换器会为各区块分配互不重叠的标识符范围。这能保留区块内的公司身份，但不意味着相同标识符跨区块对应同一家已知公司。
生成的文件会将代码、时间戳和数据拆分字段与特征一同保留，转换器还会将各拆分输出合并为一个全量数据文件，并生成数据概况。脚本也提供下载、文件验证和转换选项。其证据属于结构性描述，而非分析结果：它说明预期输入和输出，但没有呈现资产定价结果、预测评估或投资策略。文件大小检查允许一定偏差，不应将匿名标识符解释为真实公司的映射，也不应认为它们在已发布的数据拆分之间保持连续。
核心观点
- 已发布张量中的位置提供了匿名公司标识符，并在每个数据区块内保持不变。
- 分配不同标识符范围可避免在没有已发布映射的情况下意外匹配不同区块中的公司身份。
- Parquet输出保留公司、日期和数据拆分字段，供后续分析使用。
- 该工具会检查预期文件，并可为转换后的输出生成数据概况。
- 本文档介绍数据准备，而非检验投资方法。
标签
全文
# download.py
```py
#!/usr/bin/env python3
"""
Download Chen-Pelger-Zhu (2020) academic asset pricing dataset.
This dataset contains ~1.2M stock-month observations with 46 firm characteristics
and monthly returns. The released tensors retain persistent anonymous firm axes
within each published train, validation, and test block.
Source: https://github.com/jasonzy121/Deep_Learning_Asset_Pricing
Paper: "Deep Learning in Asset Pricing" (Chen, Pelger, Zhu, 2020)
Usage:
python scripts/download_academic.py # Download all files
python scripts/download_academic.py --check # Verify existing files
python scripts/download_academic.py --force # Force re-download
Data structure:
academic/dl_asset_pricing/
├── RetChar.csv # 1.1GB - Stock returns + 46 characteristics
├── Macro.csv # 1.8MB - 178 macroeconomic indicators
├── char/ # Pre-split characteristic numpy arrays
│ ├── Char_train.npz # 1967-1986
│ ├── Char_valid.npz # 1987-1991
│ └── Char_test.npz # 1992-2016
├── macro/ # Pre-split macro numpy arrays
│ ├── macro_train.npz
│ ├── macro_valid.npz
│ └── macro_test.npz
└── RF/ # Pre-processed features for random forest
├── RF_train_normalized_task_1.npz
├── RF_valid_normalized_task_1.npz
└── RF_test_normalized_task_1.npz
Note: The original data is hosted on Google Drive via the GitHub repo.
This script downloads from the GitHub release or provides manual instructions.
"""
import argparse
import os
import sys
import zipfile
from pathlib import Path
from ml4t.data.storage.data_profile import (
generate_profile,
get_profile_path,
save_profile,
)
from utils.downloading import resolve_data_dir
# Expected files and their approximate sizes (for verification)
EXPECTED_FILES = {
"RetChar.csv": 1_152_695_892, # ~1.1GB
"Macro.csv": 1_816_127, # ~1.8MB
"char/Char_train.npz": 332_629_190,
"char/Char_valid.npz": 75_511_430,
"char/Char_test.npz": 805_509_830,
"macro/macro_train.npz": 359_850,
"macro/macro_valid.npz": 98_490,
"macro/macro_test.npz": 446_970,
"RF/RF_train_normalized_task_1.npz": 332_629_190,
"RF/RF_valid_normalized_task_1.npz": 75_511_430,
"RF/RF_test_normalized_task_1.npz": 805_509_830,
}
# Source repository
GITHUB_REPO = "https://github.com/jasonzy121/Deep_Learning_Asset_Pricing"
# Google Drive folder containing all data files
GDRIVE_FOLDER_URL = "https://drive.google.com/drive/folders/1TrYzMUA_xLID5-gXOy_as8sH2ahLwz-l"
# Google Drive file IDs (from the original repo)
# Note: These may change if the authors update the data
GDRIVE_FILES = {
"data.zip": "1nYHpJ2lNm-qDX5iq18-HaL1H6z7lPGVi", # Main data archive
}
# Additional academic data files (separate sources)
ADDITIONAL_FILES = {
"SDF-Time-Series.xlsx": {
"url": "https://www.dropbox.com/scl/fi/6wgeg4ztoi5vu680x01eq/SDF-Time-Series.xlsx?rlkey=ehy8zaz2fh6tyq43hpf64gczh&e=1&dl=1",
"size": 240_633, # ~241KB
"description": "SDF time series data (Pelger) - used for Ch14 latent factor validation",
},
}
def download_additional_files(data_dir: Path) -> None:
"""Download additional academic data files from direct URLs."""
import requests
for filename, info in ADDITIONAL_FILES.items():
output_path = data_dir / filename
if output_path.exists():
print(f" [OK] {filename} already exists")
continue
print(f" Downloading {filename} ({info['description']})...")
try:
resp = requests.get(info["url"], stream=True, allow_redirects=True, timeout=60)
resp.raise_for_status()
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "wb") as f:
for chunk in resp.iter_content(8192):
f.write(chunk)
print(f" [OK] {filename} ({output_path.stat().st_size:,} bytes)")
except Exception as e:
print(f" [FAIL] {filename}: {e}")
def download_from_gdrive(file_id: str, output_path: Path) -> bool:
"""Download a file from Google Drive using gdown."""
try:
import gdown
except ImportError:
print("ERROR: gdown not installed. Run: pip install gdown")
return False
url = f"https://drive.google.com/uc?id={file_id}"
print(f"Downloading from Google Drive: {file_id}")
print(f" -> {output_path}")
try:
gdown.download(url, str(output_path), quiet=False)
return output_path.exists()
except Exception as e:
print(f"ERROR: Download failed: {e}")
return False
def extract_zip(zip_path: Path, extract_dir: Path) -> bool:
"""Extract and flatten nested zip structure."""
print(f"Extracting: {zip_path}")
try:
with zipfile.ZipFile(zip_path, "r") as zf:
# List contents
members = zf.namelist()
print(f" Archive contains {len(members)} files")
# Extract to temp location first
temp_dir = extract_dir / "_temp_extract"
temp_dir.mkdir(parents=True, exist_ok=True)
zf.extractall(temp_dir)
# Flatten nested structure. The published archives wrap their
# contents in a single top-level directory (e.g. datasets/char/...
# or data/char/...); we want char/... directly under extract_dir.
for root, _dirs, files in os.walk(temp_dir):
root_path = Path(root)
rel_root = root_path.relative_to(temp_dir)
parts = rel_root.parts
# Skip macOS archive cruft (__MACOSX/... resource forks)
if parts and parts[0] == "__MACOSX":
continue
# Strip a redundant top-level wrapper directory
if parts and parts[0] in ("data", "datasets"):
rel_root = Path(*parts[1:]) if len(parts) > 1 else Path(".")
for file in files:
if file == ".DS_Store":
continue
src = root_path / file
if rel_root == Path("."):
dst = extract_dir / file
else:
dst = extract_dir / rel_root / file
dst.parent.mkdir(parents=True, exist_ok=True)
if not dst.exists():
src.rename(dst)
print(f" Extracted: {dst.name}")
# Cleanup temp directory
import shutil
shutil.rmtree(temp_dir, ignore_errors=True)
return True
except Exception as e:
print(f"ERROR: Extraction failed: {e}")
return False
def verify_files(data_dir: Path) -> tuple[list[str], list[str]]:
"""Verify expected files exist and have correct sizes."""
found = []
missing = []
for filename, expected_size in EXPECTED_FILES.items():
filepath = data_dir / filename
if filepath.exists():
actual_size = filepath.stat().st_size
# Allow 1% tolerance for size differences
if abs(actual_size - expected_size) / expected_size < 0.01:
found.append(filename)
else:
print(f" WARNING: {filename} size mismatch: {actual_size} vs {expected_size}")
found.append(filename) # Still count as found
else:
missing.append(filename)
return found, missing
def _parquet_has_persistent_symbols(path: Path) -> bool:
"""Return whether an existing parquet satisfies the identity-preserving schema."""
if not path.exists():
return False
import polars as pl
try:
names = set(pl.scan_parquet(path).collect_schema().names())
except Exception:
return False
return {"symbol", "timestamp", "split"}.issubset(names)
def print_manual_instructions(data_dir: Path) -> None:
"""Print manual download instructions."""
print("\n" + "=" * 70)
print("MANUAL DOWNLOAD INSTRUCTIONS")
print("=" * 70)
print(f"\nSource: {GITHUB_REPO}")
print(f"Direct: {GDRIVE_FOLDER_URL}")
print("\n1. Open the Google Drive folder link above")
print(
"2. Download all files (datasets.zip ~367MB, RetChar.csv ~1.1GB, Macro.csv, sample_checkpoints.zip)"
)
print("3. Extract zip files")
print("4. Place files in:", data_dir)
print("\nExpected structure after extraction:")
print(f" {data_dir}/")
print(" ├── RetChar.csv # Main characteristics + returns")
print(" ├── Macro.csv # Macroeconomic indicators")
print(" ├── char/ # Pre-split numpy arrays")
print(" ├── macro/ # Pre-split macro arrays")
print(" └── RF/ # Random forest features")
print("\n5. Run this script again with --check to verify")
print("=" * 70)
def main():
parser = argparse.ArgumentParser(
description="Download Chen-Pelger-Zhu (2020) academic asset pricing dataset"
)
parser.add_argument("--check", action="store_true", help="Verify existing files only")
parser.add_argument("--force", action="store_true", help="Force re-download")
parser.add_argument("--convert", action="store_true", help="Convert CSV to parquet format")
parser.add_argument(
"--data-path",
type=Path,
help="Override data directory (default: $ML4T_DATA_PATH or repo/data)",
)
args = parser.parse_args()
# Handle --convert flag
if args.convert:
data_dir = resolve_data_dir(args.data_path)
if convert_to_parquet(data_dir):
return 0
return 1
# Determine data directory
data_dir = resolve_data_dir(args.data_path)
academic_dir = data_dir / "equities" / "firm_characteristics" / "dl_asset_pricing"
print("=" * 70)
print("CHEN-PELGER-ZHU (2020) ACADEMIC DATASET")
print("=" * 70)
print(f"Target directory: {academic_dir}")
print()
# Check existing files
found, missing = verify_files(academic_dir)
print(f"Files found: {len(found)}/{len(EXPECTED_FILES)}")
if found:
for f in found:
print(f" [OK] {f}")
if missing:
print(f"\nFiles missing: {len(missing)}")
for f in missing:
print(f" [FAIL] {f}")
# If just checking, exit
if args.check:
if not missing:
print("\n[OK] All files present and verified!")
return 0
else:
print(f"\n[FAIL] Missing {len(missing)} files")
return 1
# If all source files exist and not forcing, skip download but ensure the
# parquet outputs exist (convert only if missing; the CSV read is ~1.1 GB).
if not missing and not args.force:
print("\n[OK] All source files already downloaded!")
print(" Use --force to re-download")
all_parquet = (
data_dir / "equities" / "firm_characteristics" / "firm_characteristics_all.parquet"
)
if _parquet_has_persistent_symbols(all_parquet):
return 0
if all_parquet.exists():
print(" Existing parquet predates persistent-symbol recovery; regenerating.")
return 0 if convert_to_parquet(data_dir) else 1
# Try automatic download
print("\nAttempting automatic download...")
academic_dir.mkdir(parents=True, exist_ok=True)
try:
import gdown
except ImportError:
print("gdown not available for automatic download")
print_manual_instructions(academic_dir)
return 1
# Method 1: Download entire folder (more reliable than single-file IDs)
print(f"Downloading from Google Drive folder: {GDRIVE_FOLDER_URL}")
print(" ~1.5 GB across 4 files (RetChar.csv ~1.1 GB); per-file progress below.")
try:
# NOTE: no remaining_ok kwarg; it was removed in gdown 6.x and passing it
# raises TypeError, which silently aborts the (working) folder download.
gdown.download_folder(GDRIVE_FOLDER_URL, output=str(academic_dir), quiet=False)
except Exception as e:
print(f"Folder download failed: {e}")
# Extract datasets.zip if present
datasets_zip = academic_dir / "datasets.zip"
if datasets_zip.exists():
if extract_zip(datasets_zip, academic_dir):
datasets_zip.unlink(missing_ok=True)
# Extract sample_checkpoints.zip if present
checkpoints_zip = academic_dir / "sample_checkpoints.zip"
if checkpoints_zip.exists():
if extract_zip(checkpoints_zip, academic_dir):
checkpoints_zip.unlink(missing_ok=True)
# Download additional files (SDF time series, etc.)
print("\nDownloading additional academic data files...")
download_additional_files(academic_dir)
# Verify, then convert RetChar.csv -> parquet splits
found, missing = verify_files(academic_dir)
if not missing:
print("\n[OK] Download and extraction complete!")
return 0 if convert_to_parquet(data_dir) else 1
# Method 2: Fall back to single-file download
print("\nFolder download incomplete, trying single-file download...")
zip_path = academic_dir / "data.zip"
if download_from_gdrive(GDRIVE_FILES["data.zip"], zip_path) and extract_zip(
zip_path, academic_dir
):
found, missing = verify_files(academic_dir)
if not missing:
print("\n[OK] Download and extraction complete!")
zip_path.unlink(missing_ok=True)
return 0 if convert_to_parquet(data_dir) else 1
print(f"\nWARNING: {len(missing)} files still missing after download attempts")
print_manual_instructions(academic_dir)
return 1
def _characteristic_frame(npz_path: Path, split: str, symbol_offset: int):
"""Flatten one published tensor while preserving its anonymous firm axis."""
import numpy as np
import polars as pl
with np.load(npz_path) as archive:
dates = archive["date"]
variables = [str(name) for name in archive["variable"]]
data = archive["data"]
if data.ndim != 3 or data.shape[0] != len(dates) or data.shape[2] != len(variables):
raise ValueError(f"Unexpected characteristic tensor shape in {npz_path}: {data.shape}")
if not variables or variables[0] != "ret":
raise ValueError(f"Expected 'ret' as the first variable in {npz_path}: {variables[:3]}")
valid = data[:, :, 0] != -99.99
date_index, firm_index = np.nonzero(valid)
values = data[valid]
frame = pl.DataFrame(values, schema=variables, orient="row")
return frame.with_columns(
pl.Series("symbol", symbol_offset + firm_index, dtype=pl.UInt32),
pl.Series("timestamp", dates[date_index].astype(str)).str.to_date("%Y%m%d"),
pl.lit(split).alias("split"),
).select("symbol", "timestamp", *variables, "split")
def convert_to_parquet(data_dir: Path) -> bool:
"""Convert the published tensors to canonical, identity-preserving Parquet files.
The CSV omits firm identifiers, but each NPZ block has a fixed anonymous firm
axis. Axis positions are persistent within a block. The archive publishes no
mapping between blocks, so offsets keep their identifier namespaces disjoint.
"""
import polars as pl
dl_dir = data_dir / "equities" / "firm_characteristics" / "dl_asset_pricing"
char_dir = dl_dir / "char"
output_dir = data_dir / "equities" / "firm_characteristics"
split_specs = (
("train", char_dir / "Char_train.npz", 0),
("valid", char_dir / "Char_valid.npz", 1_000_000),
("test", char_dir / "Char_test.npz", 2_000_000),
)
missing = [path for _, path, _ in split_specs if not path.exists()]
if missing:
print("ERROR: Required characteristic tensors are missing:")
for path in missing:
print(f" {path}")
return False
print("\nConverting identity-preserving characteristic tensors to parquet...")
output_dir.mkdir(parents=True, exist_ok=True)
split_paths: list[Path] = []
split_counts: dict[str, int] = {}
for split, npz_path, symbol_offset in split_specs:
print(f" Reading {npz_path}...")
frame = _characteristic_frame(npz_path, split, symbol_offset)
path = output_dir / f"firm_characteristics_{split}.parquet"
frame.write_parquet(path)
split_paths.append(path)
split_counts[split] = len(frame)
print(f" {split}: {len(frame):,} rows ({path.stat().st_size / 1e6:.1f} MB)")
all_path = output_dir / "firm_characteristics_all.parquet"
pl.concat([pl.scan_parquet(path) for path in split_paths]).sink_parquet(all_path)
all_count = sum(split_counts.values())
print(f" all: {all_count:,} rows ({all_path.stat().st_size / 1e6:.1f} MB)")
# The profile is written here because this converter is the only thing that produces
# what `load_firm_characteristics()` reads. `dataset_card.py` reads it back through
# `load_profile`; without this write it reports the dataset as unprofiled, which was
# true of every copy of this dataset before now.
profile = generate_profile(pl.read_parquet(all_path), source="firm_characteristics/download.py")
profile_path = get_profile_path(all_path)
save_profile(profile, profile_path)
print(f" profile: {profile_path.name}")
return True
if __name__ == "__main__":
sys.exit(main())
```在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。