Add download_data.py to download data from Yahoo Finance. Implement corresponding task and test functions. Add config parameters to config.py

Author: niklasniedermeier

src/backtest_bay/config.py

@@ -6,3 +6,9 @@
 ROOT = SRC.joinpath("..", "..").resolve()
 
 BLD = ROOT.joinpath("bld").resolve()
+
+# Configure input data
+STOCKS = ["AAPL", "MSFT"]
+START_DATES = ["2022-01-01", "2022-01-01"]
+END_DATES = ["2025-01-01", "2025-01-01"]
+INTERVALS = ["1d", "1d"]

src/backtest_bay/data/download_data.py

@@ -0,0 +1,80 @@
+from datetime import datetime
+
+import yfinance as yf
+
+
+def download_data(symbol, interval, start_date, end_date):
+    _validate_input(symbol, interval, start_date, end_date)
+    data = yf.download(symbol, start=start_date, end=end_date, interval=interval)
+    _validate_output(data, symbol, start_date, end_date, interval)
+    return data
+
+
+def _validate_input(symbol, interval, start_date, end_date):
+    _validate_symbol(symbol)
+    _validate_interval(interval)
+    _validate_date_format(start_date)
+    _validate_date_format(end_date)
+    _validate_date_range(start_date, end_date)
+
+
+def _validate_symbol(symbol):
+    is_symbol_string = isinstance(symbol, str)
+    if not is_symbol_string:
+        error_msg = "Symbol must be a non-empty string."
+        raise ValueError(error_msg)
+
+    is_symbol_existent = not yf.Ticker(symbol).history(period="1d").empty
+    if not is_symbol_existent:
+        error_msg = f"Invalid symbol: '{symbol}'. Please provide a valid ticker symbol."
+        raise ValueError(error_msg)
+
+
+def _validate_interval(interval):
+    valid_intervals = {
+        "1m",
+        "2m",
+        "5m",
+        "15m",
+        "30m",
+        "60m",
+        "90m",
+        "1h",
+        "1d",
+        "5d",
+        "1wk",
+        "1mo",
+        "3mo",
+    }
+    if interval not in valid_intervals:
+        error_msg = f"Invalid interval. Choose from {', '.join(valid_intervals)}."
+        raise ValueError(error_msg)
+
+
+def _validate_date_format(date_str):
+    try:
+        datetime.fromisoformat(date_str)
+    except ValueError as e:
+        error_msg = "Date must be in 'YYYY-MM-DD' format."
+        raise ValueError(error_msg) from e
+
+
+def _validate_date_range(start_date, end_date):
+    if start_date > end_date:
+        error_msg = "Start date must be before end date."
+        raise ValueError(error_msg)
+
+
+def _validate_output(data, symbol, start_date, end_date, interval):
+    if data.empty:
+        ticker = yf.Ticker(symbol)
+        hist = ticker.history(period="max")
+        min_date = hist.index.min().strftime("%Y-%m-%d")
+        max_date = hist.index.max().strftime("%Y-%m-%d")
+
+        error_msg = (
+            f"No data found for {symbol} between {start_date} and {end_date} "
+            f"with interval '{interval}'.\n"
+            f"Available data range for {symbol}: {min_date} to {max_date}."
+        )
+        raise ValueError(error_msg)

src/backtest_bay/data/task_download_data.py

@@ -0,0 +1,31 @@
+import pytask
+
+from backtest_bay.config import BLD, END_DATES, INTERVALS, SRC, START_DATES, STOCKS
+from backtest_bay.data.download_data import download_data
+
+dependencies = [SRC / "config.py", SRC / "data" / "download_data.py"]
+
+for index, _ in enumerate(STOCKS):
+    _id = (
+        STOCKS[index]
+        + "_"
+        + START_DATES[index]
+        + "_"
+        + END_DATES[index]
+        + "_"
+        + INTERVALS[index]
+    )
+
+    @pytask.task(id=_id)
+    def task_download_data(
+        symbol=STOCKS[index],
+        start_date=START_DATES[index],
+        end_date=END_DATES[index],
+        interval=INTERVALS[index],
+        depends_on=dependencies,
+        produces=BLD / (_id + ".pkl"),
+    ):
+        data = download_data(
+            symbol=symbol, start_date=start_date, end_date=end_date, interval=interval
+        )
+        data.to_pickle(produces)

src/backtest_bay/test_download_data.py

@@ -0,0 +1,24 @@
+import pytest
+
+from backtest_bay.data.download_data import _validate_symbol
+
+
+# Tests for _validate_symbol
+def test_validate_symbol_valid_cases():
+    assert _validate_symbol("AAPL") is None
+    assert _validate_symbol("MSFT") is None
+
+
+def test_validate_symbol_invalid_symbol():
+    with pytest.raises(ValueError, match="Invalid symbol: 'INVALID'"):
+        _validate_symbol("INVALID")
+
+
+def test_validate_symbol_empty_string():
+    with pytest.raises(ValueError, match="Invalid symbol: ''"):
+        _validate_symbol("")
+
+
+def test_validate_symbol_non_string():
+    with pytest.raises(ValueError, match="Symbol must be a non-empty string."):
+        _validate_symbol(123)