mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-27 20:42:26 +00:00
78445ce007
* round 1 * black * more black * providers a-thru-fmp * charting settings * charting backend * black * charting_settings * missing nasdaq import * nasdaq mypy * fed money measure * some linting * black * more linting * more linting * charting imports * black * pylint * refactor fred * more linting * rest of the providers * didn't want that file * more cleanup * missed * ta_helpers * router views * chart ta_class * forgot a file * forgot another file * biztoc * package builder * technical router * fred params * bump finviz * fix tmx * more cleanup * try again * currency historical * equity search * missed files in commit * fix some tests * black * fix more test cassettes * more patchy * black * linters * more tests * black * mypy * delete unused model * more linting * forgot files * more linting * make up your mind, black, and stop moving pylint for me * more black * back to where we were before black destroyed it * .......... * .................. * unused fetcher test * remove standard model validator * commodity router * treasury prices * oecd tests * finra * container cleanup * black * forgotten file * linter * remove wrapped "" * fix merge * patches * ecb test * finra cassette * finviz cassettes * nasdaq cassettes * oecd cassettes * yfinance cassettes * more test cassettes * black * merge cleanup * cboe helpers * black * fix typo --------- Co-authored-by: Henrique Joaquim <henriquecjoaquim@gmail.com> Co-authored-by: Igor Radovanovic <74266147+IgorWounds@users.noreply.github.com>
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""Helper functions for Quantitative Analysis."""
|
|
|
|
from typing import TYPE_CHECKING, Union
|
|
|
|
if TYPE_CHECKING:
|
|
from pandas import DataFrame, Series
|
|
|
|
|
|
# ruff: ignore=S310
|
|
def get_fama_raw(start_date: str, end_date: str) -> "DataFrame":
|
|
"""Get base Fama French data to calculate risk.
|
|
|
|
Returns
|
|
-------
|
|
DataFrame
|
|
A data with fama french model information
|
|
"""
|
|
# pylint: disable=import-outside-toplevel
|
|
from io import BytesIO
|
|
from urllib.request import urlopen
|
|
from zipfile import ZipFile
|
|
|
|
from pandas import read_csv, to_datetime, to_numeric
|
|
|
|
with urlopen( # nosec # noqa: S310 SIM117
|
|
"https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/ftp/F-F_Research_Data_Factors_CSV.zip"
|
|
) as url:
|
|
# Download Zipfile and create pandas DataFrame
|
|
with ZipFile(BytesIO(url.read())) as zipfile:
|
|
with zipfile.open("F-F_Research_Data_Factors.CSV") as zip_open:
|
|
df = read_csv(
|
|
zip_open,
|
|
header=0,
|
|
names=["Date", "MKT-RF", "SMB", "HML", "RF"],
|
|
skiprows=3,
|
|
)
|
|
|
|
df = df[df["Date"].apply(lambda x: len(str(x).strip()) == 6)]
|
|
df["Date"] = df["Date"].astype(str) + "01"
|
|
df["Date"] = to_datetime(df["Date"], format="%Y%m%d")
|
|
df["MKT-RF"] = to_numeric(df["MKT-RF"], downcast="float")
|
|
df["SMB"] = to_numeric(df["SMB"], downcast="float")
|
|
df["HML"] = to_numeric(df["HML"], downcast="float")
|
|
df["RF"] = to_numeric(df["RF"], downcast="float")
|
|
df["MKT-RF"] = df["MKT-RF"] / 100
|
|
df["SMB"] = df["SMB"] / 100
|
|
df["HML"] = df["HML"] / 100
|
|
df["RF"] = df["RF"] / 100
|
|
df = df.set_index("Date")
|
|
|
|
dt_start_date = to_datetime(start_date, format="%Y-%m-%d")
|
|
if dt_start_date > df.index.max():
|
|
raise ValueError(
|
|
f"Start date '{dt_start_date}' is after the last date available for Fama-French '{df.index[-1]}'"
|
|
)
|
|
|
|
df = df.loc[start_date:end_date] # type: ignore
|
|
|
|
return df
|
|
|
|
|
|
def validate_window(input_data: Union["Series", "DataFrame"], window: int) -> None:
|
|
"""Validate the window input.
|
|
|
|
Parameters
|
|
----------
|
|
input_data : Union[Series, DataFrame]
|
|
The input data to be validated.
|
|
window : int
|
|
The window to be validated.
|
|
|
|
Raises
|
|
------
|
|
ValueError
|
|
If the window is greater than the input data length.
|
|
"""
|
|
if window > len(input_data):
|
|
raise ValueError(
|
|
f"Window '{window}' is greater than the input data length '{len(input_data)}'"
|
|
)
|