sam/bitcoin-model
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Manage output files in less-janky fashion.

Browse Source
This commit is contained in:
sam
2024-11-17 17:14:44 -08:00
parent 7211ed1723
commit d2ac5749dc
2 changed files with 108 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
model.py
+106 -25
View File
@@ -1,10 +1,58 @@
import argparse
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import os
import seaborn as sns
import shutil
import warnings
from datetime import timedelta
from multiprocessing import Pool
from typing import BinaryIO
# Output
class Output:
"""
Output ensures result files get written to the right directory.
Example:
output = Output("output", "test-1")
with output.create("summary.txt") as f:
# path to f is "output/test-1/summary.txt"
"""
out_dir: str
def __init__(self, base_dir: str, name: str):
"""
Initialize the output manager. This will both fully delete any existing
output directory, and then create an empty directory for the output.
Args:
base_dir: The root directory for outputs
name: The subdir of the root, where the results files go
"""
self.out_dir = os.path.join(base_dir, name)
shutil.rmtree(self.out_dir, ignore_errors=True)
os.makedirs(self.out_dir)
def create(self, filename) -> BinaryIO:
"""
Create a new file for writing in the output directory.
"""
full_path = self.named(filename)
f = open(full_path, "w")
return f
def named(self, filename) -> str:
"""
Get the full path within the output directory for a named results file.
"""
full_path = os.path.join(self.out_dir, filename)
return full_path
# Utility functions
@@ -640,7 +688,7 @@ def analyze_bitcoin_prices(csv_path):
# Main plotting functions
def create_plots(df, start=None, end=None, project_days=365):
def create_plots(df, output: Output, start=None, end=None, project_days=365):
"""
Create enhanced plots including market maturity visualization.
"""
@@ -832,7 +880,9 @@ def create_plots(df, start=None, end=None, project_days=365):
# Save the plot
start_str = start if start else plot_df["Date"].min().strftime("%Y-%m-%d")
end_str = end if end else plot_df["Date"].max().strftime("%Y-%m-%d")
filename = f"bitcoin_analysis_{start_str}_to_{end_str}_with_projections.png"
filename = output.named(
f"bitcoin_analysis_{start_str}_to_{end_str}_with_projections.png"
)
# Use tight_layout with adjusted parameters
plt.tight_layout(pad=2.0)
@@ -842,7 +892,7 @@ def create_plots(df, start=None, end=None, project_days=365):
return projections
def visualize_cycle_patterns(df, cycle_returns, cycle_volatility):
def visualize_cycle_patterns(df, output: Output, cycle_returns, cycle_volatility):
"""
Create enhanced visualization of Bitcoin's behavior across halving cycles.
"""
@@ -1041,12 +1091,17 @@ def visualize_cycle_patterns(df, cycle_returns, cycle_volatility):
plt.tight_layout()
# Save the plot
plt.savefig("bitcoin_cycle_patterns.png", dpi=300, bbox_inches="tight")
filename = output.named("bitcoin_cycle_patterns.png")
plt.savefig(filename, dpi=300, bbox_inches="tight")
plt.close()
def create_backtest_plot(
df, backtest_date="2020-05-11", start_date="2012-11-28", project_days=1650
df,
output: Output,
backtest_date="2020-05-11",
start_date="2012-11-28",
project_days=1650,
):
"""
Create a plot comparing actual price history against model projections from a historical date.
@@ -1246,7 +1301,9 @@ def create_backtest_plot(
# Adjust layout and save
plt.tight_layout()
filename = f'bitcoin_backtest_{start_date.strftime("%Y%m%d")}_to_{backtest_date.strftime("%Y%m%d")}.png'
filename = output.named(
f'bitcoin_backtest_{start_date.strftime("%Y%m%d")}_to_{backtest_date.strftime("%Y%m%d")}.png'
)
plt.savefig(filename, dpi=300, bbox_inches="tight")
plt.close()
@@ -1254,11 +1311,11 @@ def create_backtest_plot(
def run_projection(args):
df, start = args
_ = create_plots(df, start=start, project_days=365 * 4)
df, start, output = args
_ = create_plots(df, output, start=start, project_days=365 * 4)
def run_projections(df):
def run_projections(df, output: Output):
# Create main projection
projection_starts = [
"2011-01-01",
@@ -1268,7 +1325,7 @@ def run_projections(df):
"2015-01-01",
"2016-07-09",
]
args = [(df, start) for start in projection_starts]
args = [(df, start, output) for start in projection_starts]
with Pool() as pool:
pool.map(run_projection, args)
@@ -1281,13 +1338,13 @@ def run_single_backtest(args):
Args:
args: tuple of (params dict, DataFrame)
"""
params, df = args
params, df, output = args
try:
# Create a copy of params without the description
backtest_params = params.copy()
backtest_params.pop("description", None)
projections, metrics = create_backtest_plot(df, **backtest_params)
projections, metrics = create_backtest_plot(df, output, **backtest_params)
# Ensure metrics has all required keys with default values
if metrics is None:
@@ -1317,7 +1374,9 @@ def run_single_backtest(args):
return {"params": params, "error": str(e), "success": False}
def run_systematic_backtests(df, validation_years=2, min_training_years=8):
def run_systematic_backtests(
df, output: Output, validation_years=2, min_training_years=8
):
"""
Run a comprehensive suite of backtests with consistent validation periods.
Uses sliding windows for both start and end dates.
@@ -1463,7 +1522,7 @@ def run_systematic_backtests(df, validation_years=2, min_training_years=8):
print(f"- {period['description']}")
# Create args tuples with params and DataFrame
args = [(params, df) for params in unique_periods]
args = [(params, df, output) for params in unique_periods]
# Use multiprocessing
with Pool() as pool:
@@ -1517,7 +1576,7 @@ def run_systematic_backtests(df, validation_years=2, min_training_years=8):
stress_metrics = calculate_category_metrics(stress_periods_results)
# Write detailed results
with open("bitcoin_backtest_results_summary.txt", "w") as f:
with output.create("bitcoin_backtest_results_summary.txt") as f:
f.write("Systematic Backtest Results\n")
f.write("==========================\n\n")
@@ -1601,20 +1660,38 @@ def run_systematic_backtests(df, validation_years=2, min_training_years=8):
)
# if __name__ == "__main__":
# analysis, df = analyze_bitcoin_prices("prices.csv")
# procs = []
#
# for proc in procs:
# proc.join()
#
# CLI
if __name__ == "__main__":
def get_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="model",
description="Bitcoin price model",
)
parser.add_argument(
"-o",
"--output",
help="output base directory",
default="./output",
)
parser.add_argument(
"-n",
"--name",
help="subdir of output base directory",
default="baseline",
)
return parser.parse_args()
def main():
args = get_args()
global output
output = Output(args.output, args.name)
analysis, df = analyze_bitcoin_prices("prices.csv")
run_projections(df)
run_projections(df, output)
normal_metrics, stress_metrics, normal_results, stress_results, failed_tests = (
run_systematic_backtests(df)
run_systematic_backtests(df, output)
)
print("\nAggregate Metrics:")
@@ -1631,3 +1708,7 @@ if __name__ == "__main__":
for test in failed_tests:
print(f"Period: {test['params']['description']}")
print(f"Error: {test['error']}\n")
if __name__ == "__main__":
main()