Yuxuan Yan
528620b271
- pyproject.toml: fail_under 80 → 95 - test_alpha: +79 lines - test_cli_workflow: +226 lines - test_derived: +121 lines - test_downloader_contracts: +169 lines - test_features: +16 lines - test_minute_downloader: +81 lines - test_portfolio: +208 lines
772 lines
27 KiB
Python
772 lines
27 KiB
Python
"""CLI handoff tests for the offline daily workflow."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
import click
|
|
import pandas as pd
|
|
from click.testing import CliRunner
|
|
|
|
from cli import cli
|
|
import pipeline.derived.cli as derived_cli
|
|
import pipeline.features.cli as features_cli
|
|
from tests.helpers import (
|
|
make_generated_daily_bars,
|
|
make_generated_derived_features,
|
|
make_generated_minute_bars,
|
|
)
|
|
|
|
|
|
FIXTURE_PATH = Path(__file__).parent / "fixtures" / "daily_bars_real_2024_01_sample.pq"
|
|
|
|
|
|
def _invoke_ok(runner: CliRunner, args: list[str]):
|
|
result = runner.invoke(cli, args)
|
|
assert result.exit_code == 0, result.output
|
|
return result
|
|
|
|
|
|
def _invoke_error(runner: CliRunner, args: list[str]):
|
|
result = runner.invoke(cli, args)
|
|
assert result.exit_code != 0, result.output
|
|
return result
|
|
|
|
|
|
def test_cli_daily_workflow_handoffs_stay_in_tmp_path(tmp_path):
|
|
runner = CliRunner()
|
|
daily_bars = make_generated_daily_bars()
|
|
minute_bars = make_generated_minute_bars(daily_bars)
|
|
derived_features = make_generated_derived_features(daily_bars)
|
|
|
|
daily_path = tmp_path / "daily_bars.pq"
|
|
minute_path = tmp_path / "minute_bars.pq"
|
|
derived_input_path = tmp_path / "derived_input.pq"
|
|
daily_bars.to_parquet(daily_path, index=False)
|
|
minute_bars.to_parquet(minute_path, index=False)
|
|
derived_features.to_parquet(derived_input_path, index=False)
|
|
|
|
ingest_dir = tmp_path / "derived_ingested"
|
|
ingest_result = _invoke_ok(runner, [
|
|
"derived", "ingest",
|
|
"--input-path", str(derived_input_path),
|
|
"--derived-name", "toy_features",
|
|
"--output-dir", str(ingest_dir),
|
|
])
|
|
ingested_feature_path = ingest_dir / "toy_features.pq"
|
|
assert "Saved derived data:" in ingest_result.output
|
|
assert ingested_feature_path.exists()
|
|
|
|
validate_result = _invoke_ok(runner, [
|
|
"derived", "validate",
|
|
"--input-path", str(ingested_feature_path),
|
|
])
|
|
assert "Valid derived data:" in validate_result.output
|
|
assert "rows" in validate_result.output
|
|
|
|
computed_derived_dir = tmp_path / "derived_computed"
|
|
derived_compute_result = _invoke_ok(runner, [
|
|
"derived", "compute",
|
|
"--daily-path", str(daily_path),
|
|
"--minute-path", str(minute_path),
|
|
"--derived-type", "minute_daily_summary",
|
|
"--derived-name", "minute_summary",
|
|
"--output-dir", str(computed_derived_dir),
|
|
])
|
|
minute_summary_path = computed_derived_dir / "minute_summary.pq"
|
|
assert "Loaded daily data:" in derived_compute_result.output
|
|
assert "Loaded minute bars:" in derived_compute_result.output
|
|
assert "Saved derived data:" in derived_compute_result.output
|
|
assert minute_summary_path.exists()
|
|
assert "minute_vwap" in pd.read_parquet(minute_summary_path).columns
|
|
|
|
alpha_module_path = tmp_path / "cli_feature_alpha.py"
|
|
alpha_module_path.write_text(textwrap.dedent("""
|
|
import pandas as pd
|
|
from pipeline.alpha.base import BaseAlpha
|
|
from pipeline.alpha.registry import register_alpha
|
|
|
|
@register_alpha
|
|
class CliFeatureAlpha(BaseAlpha):
|
|
name = "cli_feature_alpha_workflow"
|
|
|
|
def __init__(self, **kwargs):
|
|
self.kwargs = kwargs
|
|
|
|
def signal_from_data(
|
|
self,
|
|
data: pd.DataFrame,
|
|
close: pd.DataFrame,
|
|
) -> pd.DataFrame:
|
|
signal = data.pivot_table(
|
|
index="date",
|
|
columns="symbol_id",
|
|
values="minute_intraday_return",
|
|
aggfunc="first",
|
|
)
|
|
fallback = close.pct_change(1, fill_method=None)
|
|
feature_signal = signal.reindex(index=close.index, columns=close.columns)
|
|
toy_signal = data.pivot_table(
|
|
index="date",
|
|
columns="symbol_id",
|
|
values="toy_feature",
|
|
aggfunc="first",
|
|
)
|
|
toy_signal = toy_signal.reindex(index=close.index, columns=close.columns)
|
|
return feature_signal.fillna(fallback) + toy_signal / 1000.0
|
|
"""))
|
|
|
|
alpha_dir = tmp_path / "alphas"
|
|
alpha_result = _invoke_ok(runner, [
|
|
"alpha", "compute",
|
|
"--data-path", str(daily_path),
|
|
"--feature-path", str(minute_summary_path),
|
|
"--feature-path", str(ingested_feature_path),
|
|
"--alpha-module", str(alpha_module_path),
|
|
"--alpha-type", "cli_feature_alpha_workflow",
|
|
"--alpha-name", "cli_feature_alpha",
|
|
"--output-dir", str(alpha_dir),
|
|
])
|
|
alpha_path = alpha_dir / "cli_feature_alpha.pq"
|
|
assert "Loaded data:" in alpha_result.output
|
|
assert "Saved alpha:" in alpha_result.output
|
|
assert "Weight stats" in alpha_result.output
|
|
assert alpha_path.exists()
|
|
assert not pd.read_parquet(alpha_path).empty
|
|
|
|
alpha_report_dir = tmp_path / "alpha_reports"
|
|
alpha_eval_result = _invoke_ok(runner, [
|
|
"alpha", "eval",
|
|
"--alpha-path", str(alpha_path),
|
|
"--data-path", str(daily_path),
|
|
"--report-dir", str(alpha_report_dir),
|
|
])
|
|
alpha_report_path = alpha_report_dir / "cli_feature_alpha_eval.json"
|
|
assert "ALPHA EVALUATION" in alpha_eval_result.output
|
|
assert "Report saved:" in alpha_eval_result.output
|
|
assert alpha_report_path.exists()
|
|
|
|
combo_dir = tmp_path / "combos"
|
|
combo_result = _invoke_ok(runner, [
|
|
"combo", "combine",
|
|
"--alpha-paths", f"{alpha_path},{alpha_path}",
|
|
"--combo-name", "cli_combo",
|
|
"--method", "equal_weight",
|
|
"--output-dir", str(combo_dir),
|
|
])
|
|
combo_path = combo_dir / "cli_combo.pq"
|
|
assert "Saved combo:" in combo_result.output
|
|
assert "Weight stats" in combo_result.output
|
|
assert combo_path.exists()
|
|
|
|
portfolio_dir = tmp_path / "portfolio"
|
|
build_result = _invoke_ok(runner, [
|
|
"portfolio", "build",
|
|
"--weights-path", str(combo_path),
|
|
"--data-path", str(daily_path),
|
|
"--booksize", "2000000",
|
|
"--portfolio-name", "cli_portfolio",
|
|
"--output-dir", str(portfolio_dir),
|
|
])
|
|
positions_path = portfolio_dir / "cli_portfolio.pq"
|
|
assert "Saved positions:" in build_result.output
|
|
assert "Gross exposure" in build_result.output
|
|
assert positions_path.exists()
|
|
|
|
execution_dir = tmp_path / "execution"
|
|
simulate_result = _invoke_ok(runner, [
|
|
"portfolio", "simulate",
|
|
"--positions-path", str(positions_path),
|
|
"--data-path", str(daily_path),
|
|
"--constraint", "suspension",
|
|
"--constraint", "price_limit",
|
|
"--constraint", "volume_cap",
|
|
"--cost-bps", "5",
|
|
"--slippage-bps", "5",
|
|
"--volume-frac", "0.02",
|
|
"--output-dir", str(execution_dir),
|
|
])
|
|
fills_path = execution_dir / "fills" / "cli_portfolio.pq"
|
|
pnl_path = execution_dir / "pnl" / "cli_portfolio.pq"
|
|
assert "Saved fills:" in simulate_result.output
|
|
assert "Saved pnl:" in simulate_result.output
|
|
assert "Total PnL:" in simulate_result.output
|
|
assert fills_path.exists()
|
|
assert pnl_path.exists()
|
|
|
|
eval_result = _invoke_ok(runner, [
|
|
"portfolio", "eval",
|
|
"--positions-path", str(positions_path),
|
|
"--data-path", str(daily_path),
|
|
])
|
|
assert "Research-portfolio metrics:" in eval_result.output
|
|
assert "cumulative_return" in eval_result.output
|
|
assert "fitness" in eval_result.output
|
|
|
|
pqcat_result = _invoke_ok(runner, [
|
|
"pqcat",
|
|
str(positions_path),
|
|
"--info",
|
|
])
|
|
assert "shape:" in pqcat_result.output
|
|
assert "dtypes:" in pqcat_result.output
|
|
assert "position_shares" in pqcat_result.output
|
|
|
|
alphaview_result = _invoke_ok(runner, [
|
|
"alphaview",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-path", str(alpha_path),
|
|
"--symbol", "sh600000",
|
|
"--start-date", "2024-01-02",
|
|
"--end-date", "2024-01-12",
|
|
"--columns", "close,volume",
|
|
])
|
|
assert "symbol: sh600000" in alphaview_result.output
|
|
assert "cli_feature_alpha" in alphaview_result.output
|
|
|
|
|
|
def test_cli_pipeline_accepts_partitioned_daily_dataset(tmp_path):
|
|
runner = CliRunner()
|
|
daily_bars = make_generated_daily_bars(include_missing=False)
|
|
dataset_dir = tmp_path / "daily_dataset"
|
|
dataset_frame = daily_bars.copy()
|
|
dataset_frame["month"] = dataset_frame["date"].dt.strftime("%Y-%m")
|
|
dataset_frame.to_parquet(dataset_dir, partition_cols=["month"], index=False)
|
|
|
|
alpha_dir = tmp_path / "alphas"
|
|
alpha_result = _invoke_ok(runner, [
|
|
"alpha", "compute",
|
|
"--data-path", str(dataset_dir),
|
|
"--alpha-type", "reversal",
|
|
"--alpha-name", "dataset_reversal",
|
|
"--lookback", "3",
|
|
"--output-dir", str(alpha_dir),
|
|
])
|
|
alpha_path = alpha_dir / "dataset_reversal.pq"
|
|
assert "Loaded data:" in alpha_result.output
|
|
assert alpha_path.exists()
|
|
|
|
combo_dir = tmp_path / "combos"
|
|
_invoke_ok(runner, [
|
|
"combo", "combine",
|
|
"--alpha-paths", str(alpha_path),
|
|
"--combo-name", "dataset_combo",
|
|
"--output-dir", str(combo_dir),
|
|
])
|
|
combo_path = combo_dir / "dataset_combo.pq"
|
|
assert combo_path.exists()
|
|
|
|
portfolio_dir = tmp_path / "portfolio"
|
|
_invoke_ok(runner, [
|
|
"portfolio", "build",
|
|
"--weights-path", str(combo_path),
|
|
"--data-path", str(dataset_dir),
|
|
"--booksize", "1000000",
|
|
"--portfolio-name", "dataset_portfolio",
|
|
"--output-dir", str(portfolio_dir),
|
|
])
|
|
positions_path = portfolio_dir / "dataset_portfolio.pq"
|
|
assert positions_path.exists()
|
|
|
|
execution_dir = tmp_path / "execution"
|
|
simulate_result = _invoke_ok(runner, [
|
|
"portfolio", "simulate",
|
|
"--positions-path", str(positions_path),
|
|
"--data-path", str(dataset_dir),
|
|
"--constraint", "suspension",
|
|
"--output-dir", str(execution_dir),
|
|
])
|
|
assert "Saved fills:" in simulate_result.output
|
|
assert (execution_dir / "fills" / "dataset_portfolio.pq").exists()
|
|
assert (execution_dir / "pnl" / "dataset_portfolio.pq").exists()
|
|
|
|
|
|
def test_cli_liquid_universe_masks_to_top_liquid_names(tmp_path):
|
|
runner = CliRunner()
|
|
daily_bars = make_generated_daily_bars(n_sessions=75, include_missing=False)
|
|
daily_path = tmp_path / "daily_bars_75d.pq"
|
|
daily_bars.to_parquet(daily_path, index=False)
|
|
|
|
alpha_dir = tmp_path / "alphas"
|
|
result = _invoke_ok(runner, [
|
|
"alpha", "compute",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-type", "reversal_rank",
|
|
"--alpha-name", "liquid_rank",
|
|
"--lookback", "3",
|
|
"--liquid-universe",
|
|
"--universe-top-n", "2",
|
|
"--output-dir", str(alpha_dir),
|
|
])
|
|
|
|
alpha_path = alpha_dir / "liquid_rank.pq"
|
|
alpha = pd.read_parquet(alpha_path)
|
|
nonzero = alpha[alpha["weight"] != 0.0]
|
|
assert "Saved alpha:" in result.output
|
|
assert alpha_path.exists()
|
|
assert not nonzero.empty
|
|
assert nonzero.groupby("date")["symbol_id"].nunique().max() <= 2
|
|
|
|
|
|
def test_cli_real_fixture_round_trips_through_portfolio(tmp_path):
|
|
runner = CliRunner()
|
|
|
|
alpha_dir = tmp_path / "alphas"
|
|
_invoke_ok(runner, [
|
|
"alpha", "compute",
|
|
"--data-path", str(FIXTURE_PATH),
|
|
"--alpha-type", "reversal_vol",
|
|
"--alpha-name", "real_cli_reversal_vol",
|
|
"--lookback", "3",
|
|
"--vol-window", "3",
|
|
"--output-dir", str(alpha_dir),
|
|
])
|
|
alpha_path = alpha_dir / "real_cli_reversal_vol.pq"
|
|
assert alpha_path.exists()
|
|
assert not pd.read_parquet(alpha_path).empty
|
|
|
|
combo_dir = tmp_path / "combos"
|
|
_invoke_ok(runner, [
|
|
"combo", "combine",
|
|
"--alpha-paths", str(alpha_path),
|
|
"--combo-name", "real_cli_combo",
|
|
"--output-dir", str(combo_dir),
|
|
])
|
|
combo_path = combo_dir / "real_cli_combo.pq"
|
|
assert combo_path.exists()
|
|
|
|
portfolio_dir = tmp_path / "portfolio"
|
|
_invoke_ok(runner, [
|
|
"portfolio", "build",
|
|
"--weights-path", str(combo_path),
|
|
"--data-path", str(FIXTURE_PATH),
|
|
"--booksize", "1000000",
|
|
"--portfolio-name", "real_cli_portfolio",
|
|
"--output-dir", str(portfolio_dir),
|
|
])
|
|
positions_path = portfolio_dir / "real_cli_portfolio.pq"
|
|
positions = pd.read_parquet(positions_path)
|
|
assert not positions.empty
|
|
|
|
eval_result = _invoke_ok(runner, [
|
|
"portfolio", "eval",
|
|
"--positions-path", str(positions_path),
|
|
"--data-path", str(FIXTURE_PATH),
|
|
])
|
|
assert "Research-portfolio metrics:" in eval_result.output
|
|
|
|
|
|
def test_cli_error_paths_are_clear_for_bad_user_inputs(tmp_path):
|
|
runner = CliRunner()
|
|
daily_bars = make_generated_daily_bars()
|
|
daily_path = tmp_path / "daily_bars.pq"
|
|
daily_bars.to_parquet(daily_path, index=False)
|
|
|
|
unknown_alpha = _invoke_error(runner, [
|
|
"alpha", "compute",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-type", "does_not_exist",
|
|
"--alpha-name", "bad",
|
|
"--output-dir", str(tmp_path / "alphas"),
|
|
])
|
|
assert "Unknown alpha-type" in unknown_alpha.output
|
|
|
|
malformed_param = _invoke_error(runner, [
|
|
"alpha", "compute",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-type", "reversal",
|
|
"--alpha-name", "bad_param",
|
|
"--param", "not-an-assignment",
|
|
"--output-dir", str(tmp_path / "alphas"),
|
|
])
|
|
assert "--param must be name=value" in malformed_param.output
|
|
|
|
unknown_derived = _invoke_error(runner, [
|
|
"derived", "compute",
|
|
"--daily-path", str(daily_path),
|
|
"--derived-type", "does_not_exist",
|
|
"--derived-name", "bad",
|
|
"--output-dir", str(tmp_path / "derived"),
|
|
])
|
|
assert "Unknown derived-type" in unknown_derived.output
|
|
|
|
bad_constraint_positions = pd.DataFrame({
|
|
"symbol_id": ["sh600000"],
|
|
"date": [pd.Timestamp("2024-01-02")],
|
|
"portfolio_name": ["bad_constraint"],
|
|
"target_weight": [1.0],
|
|
"target_value": [1000.0],
|
|
"target_shares": [100.0],
|
|
"position_shares": [100],
|
|
"position_value": [1000.0],
|
|
"price": [10.0],
|
|
})
|
|
positions_path = tmp_path / "positions.pq"
|
|
bad_constraint_positions.to_parquet(positions_path, index=False)
|
|
unknown_constraint = _invoke_error(runner, [
|
|
"portfolio", "simulate",
|
|
"--positions-path", str(positions_path),
|
|
"--data-path", str(daily_path),
|
|
"--constraint", "not_a_constraint",
|
|
"--output-dir", str(tmp_path / "execution"),
|
|
])
|
|
assert isinstance(unknown_constraint.exception, KeyError)
|
|
assert "not_a_constraint" in str(unknown_constraint.exception)
|
|
|
|
pqcat_missing_column = _invoke_error(runner, [
|
|
"pqcat",
|
|
str(daily_path),
|
|
"--columns", "close,not_a_column",
|
|
])
|
|
assert "Columns not found: not_a_column" in pqcat_missing_column.output
|
|
|
|
alphaview_missing_symbol = _invoke_error(runner, [
|
|
"alphaview",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-path", str(positions_path),
|
|
"--symbol", "sh999999",
|
|
])
|
|
assert "Symbol 'sh999999' not found" in alphaview_missing_symbol.output
|
|
|
|
alphaview_missing_column = _invoke_error(runner, [
|
|
"alphaview",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-path", str(positions_path),
|
|
"--symbol", "sh600000",
|
|
"--columns", "close,missing_bar_col",
|
|
])
|
|
assert "Bar columns not found: missing_bar_col" in alphaview_missing_column.output
|
|
|
|
alphaview_alpha = pd.DataFrame({
|
|
"symbol_id": ["sh600000"],
|
|
"date": [daily_bars["date"].min()],
|
|
"alpha_name": ["toy_alpha"],
|
|
"weight": [1.0],
|
|
})
|
|
alphaview_alpha_path = tmp_path / "alphaview_alpha.pq"
|
|
alphaview_alpha.to_parquet(alphaview_alpha_path, index=False)
|
|
alphaview_empty_range = _invoke_error(runner, [
|
|
"alphaview",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-path", str(alphaview_alpha_path),
|
|
"--symbol", "sh600000",
|
|
"--start-date", "2030-01-01",
|
|
])
|
|
assert "No rows in the requested date range" in alphaview_empty_range.output
|
|
|
|
empty_combo_paths = _invoke_ok(runner, [
|
|
"combo", "combine",
|
|
"--alpha-paths", " , ",
|
|
"--combo-name", "empty",
|
|
"--output-dir", str(tmp_path / "combos"),
|
|
])
|
|
assert "requires at least 1 path" in empty_combo_paths.output
|
|
|
|
|
|
def test_cli_parser_helpers_cover_string_coercion_and_bad_params():
|
|
assert derived_cli._parse_params(("n=7", "scale=2.5", "label=demo")) == {
|
|
"n": 7,
|
|
"scale": 2.5,
|
|
"label": "demo",
|
|
}
|
|
assert features_cli._parse_params(("n=7", "scale=2.5", "label=demo")) == {
|
|
"n": 7,
|
|
"scale": 2.5,
|
|
"label": "demo",
|
|
}
|
|
|
|
for module in (derived_cli, features_cli):
|
|
try:
|
|
module._parse_params(("not-an-assignment",))
|
|
except click.BadParameter as exc:
|
|
assert "--param must be name=value" in str(exc)
|
|
else:
|
|
raise AssertionError("expected BadParameter")
|
|
|
|
|
|
def test_cli_list_and_legacy_feature_paths(tmp_path):
|
|
runner = CliRunner()
|
|
daily_bars = make_generated_daily_bars(n_sessions=3, include_missing=False)
|
|
minute_bars = make_generated_minute_bars(daily_bars)
|
|
daily_path = tmp_path / "daily_bars.pq"
|
|
minute_path = tmp_path / "minute_bars.pq"
|
|
daily_bars.to_parquet(daily_path, index=False)
|
|
minute_bars.to_parquet(minute_path, index=False)
|
|
|
|
derived_list = _invoke_ok(runner, ["derived", "list"])
|
|
feature_list = _invoke_ok(runner, ["feature", "list"])
|
|
assert "minute_daily_summary" in derived_list.output
|
|
assert "minute_daily_summary" in feature_list.output
|
|
|
|
feature_dir = tmp_path / "features"
|
|
feature_compute = _invoke_ok(runner, [
|
|
"feature",
|
|
"compute",
|
|
"--minute-path",
|
|
str(minute_path),
|
|
"--daily-path",
|
|
str(daily_path),
|
|
"--feature-type",
|
|
"minute_daily_summary",
|
|
"--feature-name",
|
|
"legacy_summary",
|
|
"--output-dir",
|
|
str(feature_dir),
|
|
])
|
|
feature_path = feature_dir / "legacy_summary.pq"
|
|
assert "Loaded minute bars:" in feature_compute.output
|
|
assert "Loaded daily data:" in feature_compute.output
|
|
assert "Saved feature:" in feature_compute.output
|
|
assert feature_path.exists()
|
|
|
|
no_input = _invoke_error(runner, [
|
|
"derived",
|
|
"compute",
|
|
"--derived-type",
|
|
"minute_daily_summary",
|
|
"--derived-name",
|
|
"missing_inputs",
|
|
"--output-dir",
|
|
str(tmp_path / "derived"),
|
|
])
|
|
assert "At least one of --daily-path or --minute-path is required" in no_input.output
|
|
|
|
missing_minute = _invoke_error(runner, [
|
|
"derived",
|
|
"compute",
|
|
"--daily-path",
|
|
str(daily_path),
|
|
"--derived-type",
|
|
"minute_daily_summary",
|
|
"--derived-name",
|
|
"daily_only",
|
|
"--output-dir",
|
|
str(tmp_path / "derived"),
|
|
])
|
|
assert "minute_daily_summary requires minute input" in missing_minute.output
|
|
|
|
malformed_feature_param = _invoke_error(runner, [
|
|
"feature",
|
|
"compute",
|
|
"--minute-path",
|
|
str(minute_path),
|
|
"--feature-type",
|
|
"minute_daily_summary",
|
|
"--feature-name",
|
|
"bad_param",
|
|
"--param",
|
|
"not-an-assignment",
|
|
"--output-dir",
|
|
str(tmp_path / "features_bad"),
|
|
])
|
|
assert "--param must be name=value" in malformed_feature_param.output
|
|
|
|
unknown_feature = _invoke_error(runner, [
|
|
"feature",
|
|
"compute",
|
|
"--minute-path",
|
|
str(minute_path),
|
|
"--feature-type",
|
|
"does_not_exist",
|
|
"--feature-name",
|
|
"bad_feature",
|
|
"--output-dir",
|
|
str(tmp_path / "features_unknown"),
|
|
])
|
|
assert "Unknown feature-type" in unknown_feature.output
|
|
|
|
|
|
def test_cli_shortcuts_and_external_module_loading(tmp_path):
|
|
runner = CliRunner()
|
|
daily_bars = make_generated_daily_bars(n_sessions=8, include_missing=False)
|
|
minute_bars = make_generated_minute_bars(daily_bars)
|
|
daily_path = tmp_path / "daily_bars.pq"
|
|
minute_path = tmp_path / "minute_bars.pq"
|
|
daily_bars.to_parquet(daily_path, index=False)
|
|
minute_bars.to_parquet(minute_path, index=False)
|
|
|
|
alpha_module = tmp_path / "listed_alpha.py"
|
|
alpha_module.write_text(textwrap.dedent("""
|
|
import pandas as pd
|
|
from pipeline.alpha.base import BaseAlpha
|
|
from pipeline.alpha.registry import register_alpha
|
|
|
|
@register_alpha
|
|
class ListedAlpha(BaseAlpha):
|
|
name = "listed_alpha_cli"
|
|
|
|
def __init__(self, scale: float = 1.0, label: str = "x"):
|
|
self.scale = scale
|
|
self.label = label
|
|
|
|
def signal(self, close: pd.DataFrame) -> pd.DataFrame:
|
|
return close.pct_change(1, fill_method=None) * self.scale
|
|
"""))
|
|
derived_module = tmp_path / "listed_derived.py"
|
|
derived_module.write_text(textwrap.dedent("""
|
|
import pandas as pd
|
|
from pipeline.derived.base import BaseDerivedData
|
|
from pipeline.derived.registry import register_derived
|
|
|
|
@register_derived
|
|
class ListedDerived(BaseDerivedData):
|
|
name = "listed_derived_cli"
|
|
|
|
def __init__(self, scale: float = 1.0):
|
|
self.scale = scale
|
|
|
|
def compute(self, daily=None, minute=None) -> pd.DataFrame:
|
|
out = daily[["symbol_id", "date", "close"]].copy()
|
|
out["listed_value"] = out.pop("close") * self.scale
|
|
return out
|
|
"""))
|
|
derived_compute_module = tmp_path / "computed_derived.py"
|
|
derived_compute_module.write_text(textwrap.dedent("""
|
|
import pandas as pd
|
|
from pipeline.derived.base import BaseDerivedData
|
|
from pipeline.derived.registry import register_derived
|
|
|
|
@register_derived
|
|
class ComputedDerived(BaseDerivedData):
|
|
name = "computed_derived_cli"
|
|
|
|
def __init__(self, scale: float = 1.0):
|
|
self.scale = scale
|
|
|
|
def compute(self, daily=None, minute=None) -> pd.DataFrame:
|
|
out = daily[["symbol_id", "date", "close"]].copy()
|
|
out["computed_value"] = out.pop("close") * self.scale
|
|
return out
|
|
"""))
|
|
feature_module = tmp_path / "listed_feature.py"
|
|
feature_module.write_text(textwrap.dedent("""
|
|
import pandas as pd
|
|
from pipeline.features.base import BaseFeature
|
|
from pipeline.features.registry import register_feature
|
|
|
|
@register_feature
|
|
class ListedFeature(BaseFeature):
|
|
name = "listed_feature_cli"
|
|
|
|
def compute(self, daily=None, minute=None) -> pd.DataFrame:
|
|
out = minute[["symbol_id", "date", "close"]].copy()
|
|
out["date"] = pd.to_datetime(out["date"]).dt.normalize()
|
|
out = out.groupby(["symbol_id", "date"], as_index=False)["close"].mean()
|
|
return out.rename(columns={"close": "listed_feature_value"})
|
|
"""))
|
|
feature_compute_module = tmp_path / "computed_feature.py"
|
|
feature_compute_module.write_text(textwrap.dedent("""
|
|
import pandas as pd
|
|
from pipeline.features.base import BaseFeature
|
|
from pipeline.features.registry import register_feature
|
|
|
|
@register_feature
|
|
class ComputedFeature(BaseFeature):
|
|
name = "computed_feature_cli"
|
|
|
|
def compute(self, daily=None, minute=None) -> pd.DataFrame:
|
|
out = minute[["symbol_id", "date", "close"]].copy()
|
|
out["date"] = pd.to_datetime(out["date"]).dt.normalize()
|
|
out = out.groupby(["symbol_id", "date"], as_index=False)["close"].mean()
|
|
return out.rename(columns={"close": "computed_feature_value"})
|
|
"""))
|
|
|
|
alpha_list = _invoke_ok(runner, [
|
|
"alpha", "list",
|
|
"--alpha-module", str(alpha_module),
|
|
])
|
|
assert "listed_alpha_cli" in alpha_list.output
|
|
|
|
alpha_dir = tmp_path / "alphas"
|
|
reversal = _invoke_ok(runner, [
|
|
"alpha", "reversal",
|
|
"--data-path", str(daily_path),
|
|
"--output-dir", str(alpha_dir),
|
|
"--lookback", "3",
|
|
])
|
|
reversal_vol = _invoke_ok(runner, [
|
|
"alpha", "reversal-vol",
|
|
"--data-path", str(daily_path),
|
|
"--output-dir", str(alpha_dir),
|
|
"--lookback", "3",
|
|
"--vol-window", "3",
|
|
])
|
|
external_alpha = _invoke_ok(runner, [
|
|
"alpha", "compute",
|
|
"--data-path", str(daily_path),
|
|
"--alpha-type", "listed_alpha_cli",
|
|
"--alpha-name", "listed_alpha_run",
|
|
"--param", "scale=2.5",
|
|
"--param", "label=demo",
|
|
"--output-dir", str(alpha_dir),
|
|
])
|
|
|
|
assert "Saved alpha:" in reversal.output
|
|
assert "Saved alpha:" in reversal_vol.output
|
|
assert "Saved alpha:" in external_alpha.output
|
|
assert (alpha_dir / "reversal_3d.pq").exists()
|
|
assert (alpha_dir / "reversal_vol_3d_3d.pq").exists()
|
|
assert (alpha_dir / "listed_alpha_run.pq").exists()
|
|
|
|
derived_list = _invoke_ok(runner, [
|
|
"derived", "list",
|
|
"--derived-module", str(derived_module),
|
|
])
|
|
assert "listed_derived_cli" in derived_list.output
|
|
derived_dir = tmp_path / "derived_external"
|
|
derived_compute = _invoke_ok(runner, [
|
|
"derived", "compute",
|
|
"--daily-path", str(daily_path),
|
|
"--derived-module", str(derived_compute_module),
|
|
"--derived-type", "computed_derived_cli",
|
|
"--derived-name", "listed_derived_run",
|
|
"--param", "scale=3",
|
|
"--output-dir", str(derived_dir),
|
|
])
|
|
assert "Saved derived data:" in derived_compute.output
|
|
assert (derived_dir / "listed_derived_run.pq").exists()
|
|
|
|
feature_list = _invoke_ok(runner, [
|
|
"feature", "list",
|
|
"--feature-module", str(feature_module),
|
|
])
|
|
assert "listed_feature_cli" in feature_list.output
|
|
feature_dir = tmp_path / "features_external"
|
|
feature_compute = _invoke_ok(runner, [
|
|
"feature", "compute",
|
|
"--minute-path", str(minute_path),
|
|
"--feature-module", str(feature_compute_module),
|
|
"--feature-type", "computed_feature_cli",
|
|
"--feature-name", "listed_feature_run",
|
|
"--output-dir", str(feature_dir),
|
|
])
|
|
assert "Saved feature:" in feature_compute.output
|
|
assert (feature_dir / "listed_feature_run.pq").exists()
|
|
|
|
|
|
def test_cli_pqcat_row_modes(tmp_path):
|
|
runner = CliRunner()
|
|
daily_bars = make_generated_daily_bars(n_sessions=3, include_missing=False)
|
|
daily_path = tmp_path / "daily_bars.pq"
|
|
daily_bars.to_parquet(daily_path, index=False)
|
|
|
|
head_result = _invoke_ok(runner, [
|
|
"pqcat",
|
|
str(daily_path),
|
|
"--head",
|
|
"2",
|
|
"--columns",
|
|
"symbol_id,close",
|
|
])
|
|
tail_result = _invoke_ok(runner, [
|
|
"pqcat",
|
|
str(daily_path),
|
|
"--tail",
|
|
"1",
|
|
])
|
|
|
|
assert "symbol_id" in head_result.output
|
|
assert "close" in head_result.output
|
|
assert "date" in tail_result.output
|