assignment-7/main.py at main · finm-python-for-finance/assignment-7 · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Dict, List

import pandas as pd

from data_loader import LoadResult, benchmark_ingestion
from metrics import (
    RollingMetricsResult,
    compute_rolling_metrics_pandas,
    compute_rolling_metrics_polars,
)
from parallel import (
    ParallelResult,
    multiprocessing_rolling_metrics_pandas,
    multiprocessing_rolling_metrics_polars,
    threaded_rolling_metrics_pandas,
    threaded_rolling_metrics_polars,
)
from portfolio import compute_portfolio_parallel, compute_portfolio_sequential, load_portfolio_structure
from reporting import build_performance_summary, create_visualizations


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Parallel analytics for financial time-series.")
    parser.add_argument("--data", type=str, default="market_data-1.csv", help="Path to market data CSV file.")
    parser.add_argument(
        "--portfolio",
        type=str,
        default="portfolio_structure-1.json",
        help="Path to portfolio structure JSON file.",
    )
    parser.add_argument("--window", type=int, default=20, help="Rolling window size.")
    parser.add_argument(
        "--report-dir",
        type=str,
        default="reports",
        help="Directory where visualizations will be written.",
    )
    return parser.parse_args()


def _compute_rolling_metrics(
    ingestion_results: Dict[str, LoadResult],
    window: int,
) -> List[RollingMetricsResult]:
    results: List[RollingMetricsResult] = []
    pandas_frame = ingestion_results["pandas"].frame
    results.append(compute_rolling_metrics_pandas(pandas_frame, window=window))

    polars_result = ingestion_results.get("polars")
    if polars_result is not None:
        try:
            results.append(compute_rolling_metrics_polars(polars_result.frame, window=window))
        except ImportError:
            pass

    return results


def _compute_parallel_runs(
    ingestion_results: Dict[str, LoadResult],
    window: int,
) -> List[ParallelResult]:
    runs: List[ParallelResult] = []

    pandas_frame: pd.DataFrame = ingestion_results["pandas"].frame
    runs.append(threaded_rolling_metrics_pandas(pandas_frame, window=window))
    runs.append(multiprocessing_rolling_metrics_pandas(pandas_frame, window=window))

    polars_result = ingestion_results.get("polars")
    if polars_result is not None:
        try:
            runs.append(threaded_rolling_metrics_polars(polars_result.frame, window=window))
            runs.append(multiprocessing_rolling_metrics_polars(polars_result.frame, window=window))
        except ImportError:
            pass

    return runs


def main() -> None:
    args = parse_args()

    ingestion_results = benchmark_ingestion(args.data)

    rolling_results = _compute_rolling_metrics(ingestion_results, window=args.window)
    parallel_results = _compute_parallel_runs(ingestion_results, window=args.window)

    portfolio_structure = load_portfolio_structure(args.portfolio)
    pandas_frame: pd.DataFrame = ingestion_results["pandas"].frame
    portfolio_sequential = compute_portfolio_sequential(portfolio_structure, pandas_frame, window=args.window)
    portfolio_parallel = compute_portfolio_parallel(portfolio_structure, pandas_frame, window=args.window)

    performance_summary = build_performance_summary(ingestion_results, rolling_results, parallel_results)
    visualization_paths = create_visualizations(performance_summary, args.report_dir)

    output = {
        "performance_summary": performance_summary.to_dict(orient="records"),
        "portfolio_sequential": portfolio_sequential,
        "portfolio_parallel": portfolio_parallel,
        "charts": {k: str(v) if v else None for k, v in visualization_paths.items()},
    }

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()