Сохранение анонимных идентификаторов компаний в данных оценки активов
Код Machine Learning for Trading
Сводка
Эта утилита для работы с данными предназначена для набора данных оценки активов 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 по оригиналу и не является его копией.