-"""Date related utilities."""
+"""Date related utilities."""
from __future__ import annotations
@@ -147,12 +147,12 @@ Source code for openseries.datefixer
)
__all__ = [
- "date_fix",
- "date_offset_foll",
- "generate_calendar_date_range",
- "get_previous_business_day_before_today",
- "holiday_calendar",
- "offset_business_days",
+ "date_fix",
+ "date_offset_foll",
+ "generate_calendar_date_range",
+ "get_previous_business_day_before_today",
+ "holiday_calendar",
+ "offset_business_days",
]
@@ -161,7 +161,7 @@ Source code for openseries.datefixer
endyear: int,
markets: str | list[str],
) -> list[str]:
- """Return a list of holiday dates mapping to list of markets closed.
+ """Return a list of holiday dates mapping to list of markets closed.
Args:
startyear: First year (inclusive) to consider.
@@ -174,15 +174,15 @@ Source code for openseries.datefixer
Raises:
MarketsNotStringNorListStrError: If any market code is not supported by
``exchange_calendars`` or the input is not a string or list of strings.
- """
+ """
market_list = [markets] if isinstance(markets, str) else list(markets)
supported = exchcal.get_calendar_names()
if not all(m in supported for m in market_list):
msg = (
- "Argument markets must be a string market code or a list of market "
- "codes supported by exchange_calendars."
+ "Argument markets must be a string market code or a list of market "
+ "codes supported by exchange_calendars."
)
raise MarketsNotStringNorListStrError(msg)
@@ -191,7 +191,7 @@ Source code for openseries.datefixer
cal = exchcal.get_calendar(m)
cal_hols = cal.regular_holidays.holidays()
my_hols: list[str] = [
- date.date().strftime("%Y-%m-%d")
+ date.date().strftime("%Y-%m-%d")
for date in cal_hols
if (startyear <= date.date().year <= endyear)
]
@@ -205,17 +205,17 @@ Source code for openseries.datefixer
def holiday_calendar(
startyear: int,
endyear: int,
- countries: CountriesType = "SE",
+ countries: CountriesType = "SE",
markets: list[str] | str | None = None,
custom_holidays: list[str] | str | None = None,
) -> busdaycalendar:
- """Generate a business calendar.
+ """Generate a business calendar.
Args:
startyear: First year in date range generated.
endyear: Last year in date range generated.
countries: (List of) country code(s) according to ISO 3166-1 alpha-2.
- Defaults to "SE".
+ Defaults to "SE".
markets: (List of) markets code(s) supported by exchange_calendars.
custom_holidays: Argument where missing holidays can be added.
@@ -225,7 +225,7 @@ Source code for openseries.datefixer
Raises:
CountriesNotStringNorListStrError: If ``countries`` is not a supported
ISO 3166-1 alpha-2 string or a list of such strings.
- """
+ """
startyear -= 1
endyear += 1
if startyear == endyear:
@@ -243,11 +243,11 @@ Source code for openseries.datefixer
for country in countries:
staging = country_holidays(country=country, years=years)
countryholidays += list(staging)
- hols = cast("list[dt.date]", list(countryholidays))
+ hols = cast("list[dt.date]", list(countryholidays))
else:
msg = (
- "Argument countries must be a string country code or "
- "a list of string country codes according to ISO 3166-1 alpha-2."
+ "Argument countries must be a string country code or "
+ "a list of string country codes according to ISO 3166-1 alpha-2."
)
raise CountriesNotStringNorListStrError(msg)
@@ -268,7 +268,7 @@ Source code for openseries.datefixer
)
hols.extend([date_fix(fixerdate=ddate) for ddate in custom_list])
- return busdaycalendar(holidays=array(sorted(set(hols)), dtype="datetime64[D]"))
+ return busdaycalendar(holidays=array(sorted(set(hols)), dtype="datetime64[D]"))
@@ -277,7 +277,7 @@ Source code for openseries.datefixer
def date_fix(
fixerdate: DateType,
) -> dt.date:
- """Parse different date formats into datetime.date.
+ """Parse different date formats into datetime.date.
Args:
fixerdate: The data item to parse.
@@ -287,18 +287,18 @@ Source code for openseries.datefixer
Raises:
TypeError: If the provided ``fixerdate`` type is not supported.
- """
- msg = f"Unknown date format {fixerdate!s} of type {type(fixerdate)!s} encountered"
+ """
+ msg = f"Unknown date format {fixerdate!s} of type {type(fixerdate)!s} encountered"
if isinstance(fixerdate, Timestamp | dt.datetime):
return fixerdate.date()
if isinstance(fixerdate, dt.date):
return fixerdate
if isinstance(fixerdate, datetime64):
return (
- dt.datetime.strptime(str(fixerdate)[:10], "%Y-%m-%d").astimezone().date()
+ dt.datetime.strptime(str(fixerdate)[:10], "%Y-%m-%d").astimezone().date()
)
if isinstance(fixerdate, str):
- return dt.datetime.strptime(fixerdate, "%Y-%m-%d").astimezone().date()
+ return dt.datetime.strptime(fixerdate, "%Y-%m-%d").astimezone().date()
raise TypeError(msg)
@@ -308,20 +308,20 @@ Source code for openseries.datefixer
def date_offset_foll(
raw_date: DateType,
months_offset: int = 12,
- countries: CountriesType = "SE",
+ countries: CountriesType = "SE",
markets: list[str] | str | None = None,
custom_holidays: list[str] | str | None = None,
*,
adjust: bool = False,
following: bool = True,
) -> dt.date:
- """Offset dates according to a given calendar.
+ """Offset dates according to a given calendar.
Args:
raw_date: The date to offset from.
months_offset: Number of months as integer. Defaults to 12.
countries: (List of) country code(s) according to ISO 3166-1 alpha-2.
- Defaults to "SE".
+ Defaults to "SE".
markets: (List of) markets code(s) supported by exchange_calendars.
custom_holidays: Argument where missing holidays can be added.
adjust: Determines if offset should adjust for business days.
@@ -331,7 +331,7 @@ Source code for openseries.datefixer
Returns:
Offset date.
- """
+ """
raw_date = date_fix(raw_date)
month_delta = relativedelta(months=months_offset)
@@ -360,22 +360,22 @@ Source code for openseries.datefixer
[docs]
def get_previous_business_day_before_today(
today: dt.date | None = None,
- countries: CountriesType = "SE",
+ countries: CountriesType = "SE",
markets: list[str] | str | None = None,
custom_holidays: list[str] | str | None = None,
) -> dt.date:
- """Bump date backwards to find the previous business day.
+ """Bump date backwards to find the previous business day.
Args:
today: Manual input of the day from where the previous business day is found.
countries: (List of) country code(s) according to ISO 3166-1 alpha-2.
- Defaults to "SE".
+ Defaults to "SE".
markets: (List of) markets code(s) supported by exchange_calendars.
custom_holidays: Argument where missing holidays can be added.
Returns:
The previous business day.
- """
+ """
if today is None:
today = dt.datetime.now().astimezone().date()
@@ -396,11 +396,11 @@ Source code for openseries.datefixer
def offset_business_days(
ddate: dt.date,
days: int,
- countries: CountriesType = "SE",
+ countries: CountriesType = "SE",
markets: list[str] | str | None = None,
custom_holidays: list[str] | str | None = None,
) -> dt.date:
- """Bump date by business days.
+ """Bump date by business days.
It first adjusts to a valid business day and then bumps with given
number of business days from there.
@@ -411,13 +411,13 @@ Source code for openseries.datefixer
that is given.
If days is set as anything other than an integer its value is set to zero.
countries: (List of) country code(s) according to ISO 3166-1 alpha-2.
- Defaults to "SE".
+ Defaults to "SE".
markets: (List of) markets code(s) supported by exchange_calendars.
custom_holidays: Argument where missing holidays can be added.
Returns:
The new offset business day.
- """
+ """
try:
days = int(days)
except TypeError:
@@ -468,7 +468,7 @@ Source code for openseries.datefixer
idx = where(array(local_bdays) == ddate)[0]
- return cast("dt.date", local_bdays[idx[0] + days])
+ return cast("dt.date", local_bdays[idx[0] + days])
@@ -478,26 +478,26 @@ Source code for openseries.datefixer
trading_days: int,
start: dt.date | None = None,
end: dt.date | None = None,
- countries: CountriesType = "SE",
+ countries: CountriesType = "SE",
markets: list[str] | str | None = None,
custom_holidays: list[str] | str | None = None,
) -> list[dt.date]:
- """Generate a list of business day calendar dates.
+ """Generate a list of business day calendar dates.
Args:
trading_days: Number of days to generate. Must be greater than zero.
start: Date when the range starts.
end: Date when the range ends.
countries: (List of) country code(s) according to ISO 3166-1 alpha-2.
- Defaults to "SE".
+ Defaults to "SE".
markets: (List of) markets code(s) supported by exchange_calendars.
custom_holidays: Argument where missing holidays can be added.
Returns:
List of business day calendar dates.
- """
+ """
if trading_days < 1:
- msg = "Argument trading_days must be greater than zero."
+ msg = "Argument trading_days must be greater than zero."
raise TradingDaysNotAboveZeroError(msg)
if start and not end:
@@ -513,7 +513,7 @@ Source code for openseries.datefixer
tmp_range = date_range(
start=adjusted_start,
periods=trading_days * 365 // 252,
- freq="D",
+ freq="D",
)
calendar = holiday_calendar(
startyear=adjusted_start.year,
@@ -544,7 +544,7 @@ Source code for openseries.datefixer
tmp_range = date_range(
end=adjusted_end,
periods=trading_days * 365 // 252,
- freq="D",
+ freq="D",
)
calendar = holiday_calendar(
startyear=date_fix(tmp_range.tolist()[0]).year,
@@ -563,8 +563,8 @@ Source code for openseries.datefixer
]
msg = (
- "Provide exactly one of start or end date. "
- "Date range is inferred from number of trading days."
+ "Provide exactly one of start or end date. "
+ "Date range is inferred from number of trading days."
)
raise BothStartAndEndError(msg)
@@ -577,7 +577,7 @@ Source code for openseries.datefixer
markets: list[str] | str | None = None,
custom_holidays: list[str] | str | None = None,
) -> DatetimeIndex:
- """Resample timeseries frequency to business calendar month end dates.
+ """Resample timeseries frequency to business calendar month end dates.
Stubs left in place. Stubs will be aligned to the shortest stub.
@@ -591,7 +591,7 @@ Source code for openseries.datefixer
Returns:
A date range aligned to business period ends.
- """
+ """
copydata = data.copy()
copydata.index = DatetimeIndex(copydata.index)
copydata = copydata.resample(rule=freq).last()
diff --git a/docs/build/html/_modules/openseries/frame.html b/docs/build/html/_modules/openseries/frame.html
index 16a0af71..82e91883 100644
--- a/docs/build/html/_modules/openseries/frame.html
+++ b/docs/build/html/_modules/openseries/frame.html
@@ -108,7 +108,7 @@
Source code for openseries.frame
-"""The OpenFrame class."""
+"""The OpenFrame class."""
from __future__ import annotations
@@ -188,31 +188,31 @@ Source code for openseries.frame
logger = getLogger(__name__)
-__all__ = ["OpenFrame"]
+__all__ = ["OpenFrame"]
[docs]
class OpenFrame(_CommonModel[SeriesFloat]):
-
"""OpenFrame objects hold OpenTimeSeries in the list constituents.
+
"""OpenFrame objects hold OpenTimeSeries in the list constituents.
The intended use is to allow comparisons across these timeseries.
Args:
constituents: List of objects of Class OpenTimeSeries.
weights: List of weights in float format. Optional.
-
"""
+
"""
-
@field_validator("constituents")
+
@field_validator("constituents")
@classmethod
def _check_labels_unique(
cls: type[OpenFrame],
tseries: list[OpenTimeSeries],
) -> list[OpenTimeSeries]:
-
"""Pydantic validator ensuring that OpenFrame labels are unique."""
+
"""Pydantic validator ensuring that OpenFrame labels are unique."""
labls = [x.label for x in tseries]
if len(set(labls)) != len(labls):
-
msg = "TimeSeries names/labels must be unique"
+
msg = "TimeSeries names/labels must be unique"
raise LabelsNotUniqueError(msg)
return tseries
@@ -223,14 +223,14 @@
Source code for openseries.frame
constituents: list[OpenTimeSeries],
weights: list[float] | None = None,
) -> None:
- """OpenFrame objects hold OpenTimeSeries in the list constituents.
+ """OpenFrame objects hold OpenTimeSeries in the list constituents.
The intended use is to allow comparisons across these timeseries.
Args:
constituents: List of objects of Class OpenTimeSeries.
weights: List of weights in float format. Optional.
- """
+ """
copied_constituents = [ts.from_deepcopy() for ts in constituents]
super().__init__(
@@ -241,16 +241,16 @@ Source code for openseries.frame
def _set_tsdf(self: Self) -> None:
- """Set the tsdf DataFrame."""
+ """Set the tsdf DataFrame."""
if self.constituents is not None and len(self.constituents) != 0:
if len(self.constituents) == 1:
self.tsdf = self.constituents[0].tsdf.copy()
else:
self.tsdf = concat(
- [x.tsdf for x in self.constituents], axis="columns", sort=True
+ [x.tsdf for x in self.constituents], axis="columns", sort=True
)
else:
- logger.warning("OpenFrame() was passed an empty list.")
+ logger.warning("OpenFrame() was passed an empty list.")
def _coerce_result(
self: Self,
@@ -261,17 +261,17 @@ Source code for openseries.frame
data=result,
index=self.tsdf.columns,
name=name,
- dtype="float64",
+ dtype="float64",
)
[docs]
def from_deepcopy(self: Self) -> Self:
-
"""Create copy of the OpenFrame object.
+
"""Create copy of the OpenFrame object.
Returns:
An OpenFrame object.
-
"""
+
"""
return deepcopy(self)
@@ -279,16 +279,16 @@ Source code for openseries.frame
[docs]
def merge_series(
self: Self,
- how: LiteralHowMerge = "outer",
+ how: LiteralHowMerge = "outer",
) -> Self:
- """Merge index of Pandas Dataframes of the constituent OpenTimeSeries.
+ """Merge index of Pandas Dataframes of the constituent OpenTimeSeries.
Args:
- how: The Pandas merge method. Defaults to "outer".
+ how: The Pandas merge method. Defaults to "outer".
Returns:
An OpenFrame object.
- """
+ """
lvl_zero = list(self.columns_lvl_zero)
self.tsdf = reduce(
lambda left, right: merge(
@@ -306,12 +306,12 @@ Source code for openseries.frame
if self.tsdf.empty:
msg = (
- "Merging OpenTimeSeries DataFrames with "
- f"argument how={how} produced an empty DataFrame."
+ "Merging OpenTimeSeries DataFrames with "
+ f"argument how={how} produced an empty DataFrame."
)
raise MergingResultedInEmptyError(msg)
- if how == "inner":
+ if how == "inner":
for xerie in self.constituents:
xerie.tsdf = xerie.tsdf.loc[self.tsdf.index]
return self
@@ -323,7 +323,7 @@ Source code for openseries.frame
self: Self,
properties: list[LiteralFrameProps] | None = None,
) -> DataFrame:
- """Calculate chosen timeseries properties.
+ """Calculate chosen timeseries properties.
Args:
properties: The properties to calculate. Defaults to calculating all
@@ -331,7 +331,7 @@ Source code for openseries.frame
Returns:
Properties of the constituent OpenTimeSeries.
- """
+ """
if properties:
props = OpenFramePropertiesList(*properties)
prop_list = [getattr(self, x) for x in props]
@@ -339,113 +339,113 @@ Source code for openseries.frame
prop_list = [
getattr(self, x) for x in OpenFramePropertiesList.allowed_strings
]
- return cast("DataFrame", concat(prop_list, axis="columns").T)
+ return cast("DataFrame", concat(prop_list, axis="columns").T)
@property
def lengths_of_items(self: Self) -> Series[int]:
- """Number of observations of all constituents.
+ """Number of observations of all constituents.
Returns:
Number of observations of all constituents.
- """
+ """
return Series(
data=[self.tsdf[col].count() for col in self.tsdf.columns],
index=self.tsdf.columns,
- name="observations",
+ name="observations",
).astype(int)
@property
def item_count(self: Self) -> int:
- """Number of constituents.
+ """Number of constituents.
Returns:
Number of constituents.
- """
+ """
return len(self.constituents)
@property
def columns_lvl_zero(self: Self) -> list[str]:
- """Level 0 values of the MultiIndex columns in the .tsdf DataFrame.
+ """Level 0 values of the MultiIndex columns in the .tsdf DataFrame.
Returns:
Level 0 values of the MultiIndex columns in the .tsdf DataFrame.
- """
+ """
return list(self.tsdf.columns.get_level_values(0))
@property
def columns_lvl_one(self: Self) -> list[ValueType]:
- """Level 1 values of the MultiIndex columns in the .tsdf DataFrame.
+ """Level 1 values of the MultiIndex columns in the .tsdf DataFrame.
Returns:
Level 1 values of the MultiIndex columns in the .tsdf DataFrame.
- """
+ """
return list(self.tsdf.columns.get_level_values(1))
@property
def _value_types(self: Self) -> list[bool]:
- """Cached value type checks for efficiency.
+ """Cached value type checks for efficiency.
Returns:
List of booleans indicating if each column is ValueType.RTRN.
- """
+ """
return [x == ValueType.RTRN for x in self.tsdf.columns.get_level_values(1)]
@property
def first_indices(self: Self) -> Series[dt.date]:
- """The first dates in the timeseries of all constituents.
+ """The first dates in the timeseries of all constituents.
Returns:
The first dates in the timeseries of all constituents.
- """
+ """
return Series(
data=[i.first_idx for i in self.constituents],
index=self.tsdf.columns,
- name="first indices",
- dtype="datetime64[ns]",
+ name="first indices",
+ dtype="datetime64[ns]",
).dt.date
@property
def last_indices(self: Self) -> Series[dt.date]:
- """The last dates in the timeseries of all constituents.
+ """The last dates in the timeseries of all constituents.
Returns:
The last dates in the timeseries of all constituents.
- """
+ """
return Series(
data=[i.last_idx for i in self.constituents],
index=self.tsdf.columns,
- name="last indices",
- dtype="datetime64[ns]",
+ name="last indices",
+ dtype="datetime64[ns]",
).dt.date
@property
def span_of_days_all(self: Self) -> Series[int]:
- """Number of days from the first date to the last for all items in the frame.
+ """Number of days from the first date to the last for all items in the frame.
Returns:
Number of days from the first date to the last for all
items in the frame.
- """
+ """
return Series(
data=[c.span_of_days for c in self.constituents],
index=self.tsdf.columns,
- name="span of days",
+ name="span of days",
).astype(int)
[docs]
def value_to_ret(self: Self) -> Self:
-
"""Convert series of values into series of returns.
+
"""Convert series of values into series of returns.
Returns:
The returns of the values in the series.
-
"""
+
"""
returns = self.tsdf.ffill().pct_change()
returns.iloc[0] = 0
new_labels: list[ValueType] = [ValueType.RTRN] * self.item_count
arrays = cast(
-
"Any",
+
"Any",
[
self.tsdf.columns.get_level_values(0),
new_labels,
@@ -459,7 +459,7 @@
Source code for openseries.frame
[docs]
def value_to_diff(self: Self, periods: int = 1) -> Self:
-
"""Convert series of values to series of their period differences.
+
"""Convert series of values to series of their period differences.
Args:
periods: The number of periods between observations over which
@@ -467,12 +467,12 @@
Source code for openseries.frame
Returns:
An OpenFrame object.
- """
+ """
self.tsdf = self.tsdf.diff(periods=periods)
self.tsdf.iloc[0] = 0
new_labels: list[ValueType] = [ValueType.RTRN] * self.item_count
arrays = cast(
- "Any",
+ "Any",
[
self.tsdf.columns.get_level_values(0),
new_labels,
@@ -485,11 +485,11 @@ Source code for openseries.frame
[docs]
def to_cumret(self: Self) -> Self:
-
"""Convert series of returns into cumulative series of values.
+
"""Convert series of returns into cumulative series of values.
Returns:
An OpenFrame object.
-
"""
+
"""
vtypes = self._value_types
if not any(vtypes):
returns = self.tsdf.ffill().pct_change()
@@ -498,7 +498,7 @@
Source code for openseries.frame
returns = self.tsdf.copy()
returns.iloc[0] = 0
else:
- msg = "Mix of series types will give inconsistent results"
+ msg = "Mix of series types will give inconsistent results"
raise MixedValuetypesError(msg)
returns = returns.add(1.0)
@@ -506,7 +506,7 @@ Source code for openseries.frame
new_labels: list[ValueType] = [ValueType.PRICE] * self.item_count
arrays = cast(
- "Any",
+ "Any",
[
self.tsdf.columns.get_level_values(0),
new_labels,
@@ -520,24 +520,24 @@ Source code for openseries.frame
[docs]
def resample(
self: Self,
- freq: LiteralBizDayFreq | str = "BME",
+ freq: LiteralBizDayFreq | str = "BME",
) -> Self:
- """Resample the timeseries frequency.
+ """Resample the timeseries frequency.
Args:
freq: The date offset string that sets the resampled frequency.
- Defaults to "BME".
+ Defaults to "BME".
Returns:
An OpenFrame object.
- """
+ """
vtypes = self._value_types
if not any(vtypes):
value_type = ValueType.PRICE
elif all(vtypes):
value_type = ValueType.RTRN
else:
- msg = "Mix of series types will give inconsistent results"
+ msg = "Mix of series types will give inconsistent results"
raise MixedValuetypesError(msg)
self.tsdf.index = DatetimeIndex(self.tsdf.index)
@@ -561,28 +561,28 @@ Source code for openseries.frame
[docs]
def resample_to_business_period_ends(
self: Self,
- freq: LiteralBizDayFreq = "BME",
- method: LiteralPandasReindexMethod = "nearest",
+ freq: LiteralBizDayFreq = "BME",
+ method: LiteralPandasReindexMethod = "nearest",
) -> Self:
- """Resamples timeseries frequency to the business calendar month end dates.
+ """Resamples timeseries frequency to the business calendar month end dates.
Stubs left in place. Stubs will be aligned to the shortest stub.
Args:
freq: The date offset string that sets the resampled frequency.
- Defaults to "BME".
+ Defaults to "BME".
method: Controls the method used to align values across columns.
Defaults to nearest.
Returns:
An OpenFrame object.
- """
+ """
vtypes = self._value_types
if any(vtypes):
msg = (
- "Do not run resample_to_business_period_ends on return series. "
- "The operation will pick the last data point in the sparser series. "
- "It will not sum returns and therefore data will be lost."
+ "Do not run resample_to_business_period_ends on return series. "
+ "The operation will pick the last data point in the sparser series. "
+ "It will not sum returns and therefore data will be lost."
)
raise ResampleDataLossError(msg)
@@ -625,7 +625,7 @@ Source code for openseries.frame
to_date: dt.date | None = None,
periods_in_a_year_fixed: DaysInYearType | None = None,
) -> DataFrame:
- """Exponentially Weighted Moving Average Volatilities and Correlation.
+ """Exponentially Weighted Moving Average Volatilities and Correlation.
Exponentially Weighted Moving Average (EWMA) for Volatilities and
Correlation.
@@ -649,7 +649,7 @@ Source code for openseries.frame
Returns:
Series volatilities and correlation.
- """
+ """
earlier, later = self.calc_range(
months_offset=months_from_last,
from_dt=from_date,
@@ -658,7 +658,7 @@ Source code for openseries.frame
if periods_in_a_year_fixed is None:
fraction = (later - earlier).days / 365.25
how_many = (
- self.tsdf.loc[cast("Timestamp", earlier) : cast("Timestamp", later)]
+ self.tsdf.loc[cast("Timestamp", earlier) : cast("Timestamp", later)]
.count()
.iloc[0]
)
@@ -667,17 +667,17 @@ Source code for openseries.frame
time_factor = periods_in_a_year_fixed
corr_label = (
- cast("tuple[str, str]", self.tsdf.iloc[:, first_column].name)[0]
- + "_VS_"
- + cast("tuple[str, str]", self.tsdf.iloc[:, second_column].name)[0]
+ cast("tuple[str, str]", self.tsdf.iloc[:, first_column].name)[0]
+ + "_VS_"
+ + cast("tuple[str, str]", self.tsdf.iloc[:, second_column].name)[0]
)
cols = [
- cast("tuple[str, str]", self.tsdf.iloc[:, first_column].name)[0],
- cast("tuple[str, str]", self.tsdf.iloc[:, second_column].name)[0],
+ cast("tuple[str, str]", self.tsdf.iloc[:, first_column].name)[0],
+ cast("tuple[str, str]", self.tsdf.iloc[:, second_column].name)[0],
]
data = self.tsdf.loc[
- cast("Timestamp", earlier) : cast("Timestamp", later)
+ cast("Timestamp", earlier) : cast("Timestamp", later)
].copy()
for rtn in cols:
@@ -740,25 +740,25 @@ Source code for openseries.frame
@property
def correl_matrix(self: Self) -> DataFrame:
- """Correlation matrix.
+ """Correlation matrix.
This property returns the correlation matrix of the time series
in the frame.
Returns:
Correlation matrix of the time series in the frame.
- """
+ """
corr_matrix = (
self.tsdf.ffill()
.pct_change()
.corr(
- method="pearson",
+ method="pearson",
min_periods=1,
)
)
corr_matrix.columns = corr_matrix.columns.get_level_values(0)
corr_matrix.index = corr_matrix.index.get_level_values(0)
- corr_matrix.index.name = "Correlation"
+ corr_matrix.index.name = "Correlation"
return corr_matrix
@@ -767,30 +767,30 @@
Source code for openseries.frame
self: Self,
new_series: OpenTimeSeries,
) -> Self:
- """To add an OpenTimeSeries object.
+ """To add an OpenTimeSeries object.
Args:
new_series: The timeseries to add.
Returns:
An OpenFrame object.
- """
+ """
self.constituents += [new_series]
- self.tsdf = concat([self.tsdf, new_series.tsdf], axis="columns", sort=True)
+ self.tsdf = concat([self.tsdf, new_series.tsdf], axis="columns", sort=True)
return self
[docs]
def delete_timeseries(self: Self, lvl_zero_item: str) -> Self:
-
"""To delete an OpenTimeSeries object.
+
"""To delete an OpenTimeSeries object.
Args:
lvl_zero_item: The .tsdf column level 0 value of the timeseries to delete.
Returns:
An OpenFrame object.
-
"""
+
"""
if self.weights:
new_c, new_w = [], []
for serie, weight in zip(self.constituents, self.weights, strict=True):
@@ -803,7 +803,7 @@
Source code for openseries.frame
self.constituents = [
item for item in self.constituents if item.label != lvl_zero_item
]
- self.tsdf = self.tsdf.drop(lvl_zero_item, axis="columns", level=0)
+ self.tsdf = self.tsdf.drop(lvl_zero_item, axis="columns", level=0)
return self
@@ -813,9 +813,9 @@
Source code for openseries.frame
self: Self,
start_cut: dt.date | None = None,
end_cut: dt.date | None = None,
- where: LiteralTrunc = "both",
+ where: LiteralTrunc = "both",
) -> Self:
- """Truncate DataFrame such that all timeseries have the same time span.
+ """Truncate DataFrame such that all timeseries have the same time span.
Args:
start_cut: New first date. Optional.
@@ -825,10 +825,10 @@ Source code for openseries.frame
Returns:
An OpenFrame object.
- """
- if not start_cut and where in ["before", "both"]:
+ """
+ if not start_cut and where in ["before", "both"]:
start_cut = self.first_indices.max()
- if not end_cut and where in ["after", "both"]:
+ if not end_cut and where in ["after", "both"]:
end_cut = self.last_indices.min()
self.tsdf = self.tsdf.sort_index()
self.tsdf = self.tsdf.truncate(before=start_cut, after=end_cut)
@@ -840,16 +840,16 @@ Source code for openseries.frame
)
if len(set(self.first_indices)) != 1:
msg = (
- f"One or more constituents still "
- f"not truncated to same start dates.\n"
- f"{self.tsdf.head()}"
+ f"One or more constituents still "
+ f"not truncated to same start dates.\n"
+ f"{self.tsdf.head()}"
)
logger.warning(msg)
if len(set(self.last_indices)) != 1:
msg = (
- f"One or more constituents still "
- f"not truncated to same end dates.\n"
- f"{self.tsdf.tail()}"
+ f"One or more constituents still "
+ f"not truncated to same end dates.\n"
+ f"{self.tsdf.tail()}"
)
logger.warning(msg)
return self
@@ -864,18 +864,18 @@ Source code for openseries.frame
*,
base_zero: bool = True,
) -> None:
- """Calculate cumulative relative return between two series.
+ """Calculate cumulative relative return between two series.
Args:
long_column: Column number of timeseries bought. Defaults to 0.
short_column: Column number of timeseries sold. Defaults to 1.
base_zero: If set to False 1.0 is added to allow for a capital base and
to allow a volatility calculation. Defaults to True.
- """
+ """
rel_label = (
- cast("tuple[str, str]", self.tsdf.iloc[:, long_column].name)[0]
- + "_over_"
- + cast("tuple[str, str]", self.tsdf.iloc[:, short_column].name)[0]
+ cast("tuple[str, str]", self.tsdf.iloc[:, long_column].name)[0]
+ + "_over_"
+ + cast("tuple[str, str]", self.tsdf.iloc[:, short_column].name)[0]
)
if base_zero:
self.tsdf[rel_label, ValueType.RELRTRN] = (
@@ -900,7 +900,7 @@ Source code for openseries.frame
to_date: dt.date | None = None,
periods_in_a_year_fixed: DaysInYearType | None = None,
) -> Series[float]:
- """Tracking Error.
+ """Tracking Error.
Calculates Tracking Error which is the standard deviation of the
difference between the fund and its index returns.
@@ -919,7 +919,7 @@ Source code for openseries.frame
Returns:
Tracking Errors.
- """
+ """
earlier, later = self.calc_range(
months_offset=months_from_last,
from_dt=from_date,
@@ -948,7 +948,7 @@ Source code for openseries.frame
terrors.append(0.0)
else:
longdf = self.tsdf.loc[
- cast("Timestamp", earlier) : cast("Timestamp", later)
+ cast("Timestamp", earlier) : cast("Timestamp", later)
][item]
relative = longdf.ffill().pct_change() - shortdf_returns
vol = float(relative.std() * sqrt(time_factor))
@@ -957,8 +957,8 @@ Source code for openseries.frame
return Series(
data=terrors,
index=self.tsdf.columns,
- name=f"Tracking Errors vs {short_label}",
- dtype="float64",
+ name=f"Tracking Errors vs {short_label}",
+ dtype="float64",
)
@@ -972,7 +972,7 @@ Source code for openseries.frame
to_date: dt.date | None = None,
periods_in_a_year_fixed: DaysInYearType | None = None,
) -> Series[float]:
- """Information Ratio.
+ """Information Ratio.
The Information Ratio equals ( fund return less index return ) divided
by the Tracking Error. And the Tracking Error is the standard deviation of
@@ -991,7 +991,7 @@ Source code for openseries.frame
Returns:
Information Ratios.
- """
+ """
earlier, later = self.calc_range(
months_offset=months_from_last,
from_dt=from_date,
@@ -1020,7 +1020,7 @@ Source code for openseries.frame
ratios.append(0.0)
else:
longdf = self.tsdf.loc[
- cast("Timestamp", earlier) : cast("Timestamp", later)
+ cast("Timestamp", earlier) : cast("Timestamp", later)
][item]
relative = longdf.ffill().pct_change() - shortdf_returns
ret = float(relative.mean() * time_factor)
@@ -1030,8 +1030,8 @@ Source code for openseries.frame
return Series(
data=ratios,
index=self.tsdf.columns,
- name=f"Info Ratios vs {short_label}",
- dtype="float64",
+ name=f"Info Ratios vs {short_label}",
+ dtype="float64",
)
@@ -1041,7 +1041,7 @@ Source code for openseries.frame
mask: NDArray[bool_],
time_factor: float,
) -> float:
- """Calculate CAGR from returns array with mask.
+ """Calculate CAGR from returns array with mask.
Args:
returns_array: Returns array.
@@ -1050,7 +1050,7 @@ Source code for openseries.frame
Returns:
CAGR value.
- """
+ """
masked_array = returns_array[mask] + 1.0
if len(masked_array) == 0:
return 0.0
@@ -1066,7 +1066,7 @@ Source code for openseries.frame
down_mask: NDArray[bool_],
time_factor: float,
) -> float:
- """Calculate capture ratio for a single item.
+ """Calculate capture ratio for a single item.
Args:
ratio: Ratio type to calculate.
@@ -1081,8 +1081,8 @@ Source code for openseries.frame
Raises:
RatioInputError: If ratio is invalid.
- """
- if ratio == "up":
+ """
+ if ratio == "up":
up_rtrn = self._calculate_cagr_from_returns(
longdf_returns_np, up_mask, time_factor
)
@@ -1093,7 +1093,7 @@ Source code for openseries.frame
return 0.0
return up_rtrn / up_idx_return
- if ratio == "down":
+ if ratio == "down":
down_return = self._calculate_cagr_from_returns(
longdf_returns_np, down_mask, time_factor
)
@@ -1104,7 +1104,7 @@ Source code for openseries.frame
return 0.0
return down_return / down_idx_return
- if ratio == "both":
+ if ratio == "both":
up_rtrn = self._calculate_cagr_from_returns(
longdf_returns_np, up_mask, time_factor
)
@@ -1121,7 +1121,7 @@ Source code for openseries.frame
return 0.0
return (up_rtrn / up_idx_return) / (down_return / down_idx_return)
- msg = "ratio must be one of 'up', 'down' or 'both'."
+ msg = "ratio must be one of 'up', 'down' or 'both'."
raise RatioInputError(msg)
@@ -1135,14 +1135,14 @@
Source code for openseries.frame
to_date: dt.date | None = None,
periods_in_a_year_fixed: DaysInYearType | None = None,
) -> Series[float]:
- """Capture Ratio.
+ """Capture Ratio.
The Up (Down) Capture Ratio is calculated by dividing the CAGR
of the asset during periods that the benchmark returns are positive (negative)
by the CAGR of the benchmark during the same periods.
CaptureRatio.BOTH is the Up ratio divided by the Down ratio.
- Source: 'Capture Ratios: A Popular Method of Measuring Portfolio Performance
- in Practice', Don R. Cox and Delbert C. Goff, Journal of Economics and
+ Source: 'Capture Ratios: A Popular Method of Measuring Portfolio Performance
+ in Practice', Don R. Cox and Delbert C. Goff, Journal of Economics and
Finance Education (Vol 2 Winter 2013).
Reference: https://www.economics-finance.org/jefe/volume12-2/11ArticleCox.pdf.
@@ -1160,7 +1160,7 @@ Source code for openseries.frame
Returns:
Capture Ratios.
- """
+ """
loss_limit: float = 0.0
earlier, later = self.calc_range(
months_offset=months_from_last,
@@ -1182,7 +1182,7 @@ Source code for openseries.frame
time_factor = shortdf.count() / fraction
shortdf_returns = shortdf.ffill().pct_change()
- shortdf_returns_np = cast("NDArray[float64]", shortdf_returns.to_numpy())
+ shortdf_returns_np = cast("NDArray[float64]", shortdf_returns.to_numpy())
up_mask = shortdf_returns_np > loss_limit
down_mask = shortdf_returns_np < loss_limit
@@ -1192,10 +1192,10 @@ Source code for openseries.frame
ratios.append(0.0)
else:
longdf = self.tsdf.loc[
- cast("Timestamp", earlier) : cast("Timestamp", later)
+ cast("Timestamp", earlier) : cast("Timestamp", later)
][item]
longdf_returns = longdf.ffill().pct_change()
- longdf_returns_np = cast("NDArray[float64]", longdf_returns.to_numpy())
+ longdf_returns_np = cast("NDArray[float64]", longdf_returns.to_numpy())
ratio_value = self._calculate_capture_ratio_for_item(
ratio=ratio,
longdf_returns_np=longdf_returns_np,
@@ -1207,9 +1207,9 @@ Source code for openseries.frame
ratios.append(ratio_value)
ratio_names = {
- "up": f"Up Capture Ratios vs {short_label}",
- "down": f"Down Capture Ratios vs {short_label}",
- "both": f"Up-Down Capture Ratios vs {short_label}",
+ "up": f"Up Capture Ratios vs {short_label}",
+ "down": f"Down Capture Ratios vs {short_label}",
+ "both": f"Up-Down Capture Ratios vs {short_label}",
}
resultname = ratio_names[ratio]
@@ -1217,7 +1217,7 @@ Source code for openseries.frame
data=ratios,
index=self.tsdf.columns,
name=resultname,
- dtype="float64",
+ dtype="float64",
)
@@ -1225,9 +1225,9 @@ Source code for openseries.frame
self: Self,
column: tuple[str, ValueType] | int,
vtypes: list[bool],
- param_name: str = "column",
+ param_name: str = "column",
) -> Series[float]:
- """Extract column value based on value types.
+ """Extract column value based on value types.
Args:
column: Column reference.
@@ -1239,8 +1239,8 @@ Source code for openseries.frame
Raises:
TypeError: If column type is invalid.
- """
- msg = f"{param_name} should be a tuple[str, ValueType] or an integer."
+ """
+ msg = f"{param_name} should be a tuple[str, ValueType] or an integer."
if isinstance(column, tuple):
if all(vtypes):
return self.tsdf[column]
@@ -1259,7 +1259,7 @@ Source code for openseries.frame
market: tuple[str, ValueType] | int,
dlta_degr_freedms: int = 1,
) -> float:
- """Market Beta.
+ """Market Beta.
Calculates Beta as Co-variance of asset & market divided by Variance
of the market.
@@ -1274,14 +1274,14 @@ Source code for openseries.frame
Returns:
Beta as Co-variance of x & y divided by Variance of x.
- """
+ """
vtypes = self._value_types
if not (all(vtypes) or not any(vtypes)):
- msg = "Mix of series types will give inconsistent results"
+ msg = "Mix of series types will give inconsistent results"
raise MixedValuetypesError(msg)
- y_value = self._extract_column_value(asset, vtypes, param_name="asset")
- x_value = self._extract_column_value(market, vtypes, param_name="market")
+ y_value = self._extract_column_value(asset, vtypes, param_name="asset")
+ x_value = self._extract_column_value(market, vtypes, param_name="market")
covariance = cov(m=y_value, y=x_value, ddof=dlta_degr_freedms)
beta = covariance[0, 1] / covariance[1, 1]
@@ -1298,7 +1298,7 @@ Source code for openseries.frame
*,
fitted_series: bool = True,
) -> dict[str, float]:
- """Ordinary Least Squares fit.
+ """Ordinary Least Squares fit.
Performs a linear regression and adds a new column with a fitted line
using Ordinary Least Squares fit.
@@ -1311,30 +1311,30 @@ Source code for openseries.frame
Returns:
A dictionary with the coefficient, intercept and rsquared outputs.
- """
- msg = "y_column should be a tuple[str, ValueType] or an integer."
+ """
+ msg = "y_column should be a tuple[str, ValueType] or an integer."
if isinstance(y_column, tuple):
y_value = self.tsdf[y_column].to_numpy()
y_label = cast(
- "tuple[str, str]",
+ "tuple[str, str]",
self.tsdf[y_column].name,
)[0]
elif isinstance(y_column, int):
y_value = self.tsdf.iloc[:, y_column].to_numpy()
- y_label = cast("tuple[str, str]", self.tsdf.iloc[:, y_column].name)[0]
+ y_label = cast("tuple[str, str]", self.tsdf.iloc[:, y_column].name)[0]
else:
raise TypeError(msg)
- msg = "x_column should be a tuple[str, ValueType] or an integer."
+ msg = "x_column should be a tuple[str, ValueType] or an integer."
if isinstance(x_column, tuple):
x_value = self.tsdf[x_column].to_numpy().reshape(-1, 1)
x_label = cast(
- "tuple[str, str]",
+ "tuple[str, str]",
self.tsdf[x_column].name,
)[0]
elif isinstance(x_column, int):
x_value = self.tsdf.iloc[:, x_column].to_numpy().reshape(-1, 1)
- x_label = cast("tuple[str, str]", self.tsdf.iloc[:, x_column].name)[0]
+ x_label = cast("tuple[str, str]", self.tsdf.iloc[:, x_column].name)[0]
else:
raise TypeError(msg)
@@ -1343,9 +1343,9 @@ Source code for openseries.frame
if fitted_series:
self.tsdf[y_label, x_label] = model.predict(x_value)
return {
- "coefficient": float(model.coef_[0]),
- "intercept": float(model.intercept_),
- "rsquared": model.score(x_value, y_value),
+ "coefficient": float(model.coef_[0]),
+ "intercept": float(model.intercept_),
+ "rsquared": model.score(x_value, y_value),
}
@@ -1358,34 +1358,34 @@ Source code for openseries.frame
riskfree_rate: float = 0.0,
dlta_degr_freedms: int = 1,
) -> float:
- """Jensen's alpha.
+ """Jensen's alpha.
- The Jensen's measure, or Jensen's alpha, is a risk-adjusted performance
+ The Jensen's measure, or Jensen's alpha, is a risk-adjusted performance
measure that represents the average return on a portfolio or investment,
above or below that predicted by the capital asset pricing model (CAPM),
- given the portfolio's or investment's beta and the average market return.
+ given the portfolio's or investment's beta and the average market return.
This metric is also commonly referred to as simply alpha.
Reference: https://www.investopedia.com/terms/j/jensensmeasure.asp.
Args:
asset: The column of the asset.
- market: The column of the market against which Jensen's alpha is measured.
+ market: The column of the market against which Jensen's alpha is measured.
riskfree_rate: The return of the zero volatility riskfree asset.
Defaults to 0.0.
dlta_degr_freedms: Variance bias factor taking the value 0 or 1.
Defaults to 1.
Returns:
- Jensen's alpha.
- """
+ Jensen's alpha.
+ """
vtypes = self._value_types
if not (all(vtypes) or not any(vtypes)):
- msg = "Mix of series types will give inconsistent results"
+ msg = "Mix of series types will give inconsistent results"
raise MixedValuetypesError(msg)
- asset_rtn = self._extract_column_value(asset, vtypes, param_name="asset")
- market_rtn = self._extract_column_value(market, vtypes, param_name="market")
+ asset_rtn = self._extract_column_value(asset, vtypes, param_name="asset")
+ market_rtn = self._extract_column_value(market, vtypes, param_name="market")
asset_rtn_mean = float(asset_rtn.mean() * self.periods_in_a_year)
market_rtn_mean = float(market_rtn.mean() * self.periods_in_a_year)
@@ -1399,14 +1399,14 @@ Source code for openseries.frame
def _prepare_returns_for_portfolio(self: Self) -> DataFrame:
- """Prepare returns DataFrame for portfolio calculation.
+ """Prepare returns DataFrame for portfolio calculation.
Returns:
Returns DataFrame.
Raises:
MixedValuetypesError: If series types are mixed.
- """
+ """
vtypes = self._value_types
if not any(vtypes):
returns = self.tsdf.ffill().pct_change()
@@ -1414,32 +1414,32 @@ Source code for openseries.frame
return returns
if all(vtypes):
return self.tsdf
- msg = "Mix of series types will give inconsistent results"
+ msg = "Mix of series types will give inconsistent results"
raise MixedValuetypesError(msg)
def _calculate_eq_weights(self: Self) -> list[float]:
- """Calculate equal weights.
+ """Calculate equal weights.
Returns:
List of equal weights.
- """
+ """
return [1.0 / self.item_count] * self.item_count
def _calculate_inv_vol_weights(self: Self, returns: DataFrame) -> list[float]:
- """Calculate inverse volatility weights.
+ """Calculate inverse volatility weights.
Args:
returns: Returns DataFrame.
Returns:
List of inverse volatility weights.
- """
+ """
vol = divide(1.0, std(returns, axis=0, ddof=1))
vol[isinf(vol)] = nan
return list(divide(vol, vol.sum()))
def _calculate_max_div_weights(self: Self, returns: DataFrame) -> list[float]:
- """Calculate maximum diversification weights.
+ """Calculate maximum diversification weights.
Args:
returns: Returns DataFrame.
@@ -1450,12 +1450,12 @@ Source code for openseries.frame
Raises:
MaxDiversificationNaNError: If correlation matrix has NaN values.
MaxDiversificationNegativeWeightsError: If weights are negative.
- """
+ """
corr_matrix = corrcoef(returns.T)
corr_matrix[isinf(corr_matrix)] = nan
corr_matrix[isnan(corr_matrix)] = nan
- msga = "max_div weight strategy failed: correlation matrix contains NaN values"
+ msga = "max_div weight strategy failed: correlation matrix contains NaN values"
if isnan(corr_matrix).any():
raise MaxDiversificationNaNError(msga)
@@ -1463,29 +1463,29 @@ Source code for openseries.frame
inv_corr_sum = linalg.inv(corr_matrix).sum(axis=1)
msgb = (
- "max_div weight strategy failed: "
- "inverse correlation matrix sum contains NaN values"
+ "max_div weight strategy failed: "
+ "inverse correlation matrix sum contains NaN values"
)
if isnan(inv_corr_sum).any():
raise MaxDiversificationNaNError(msgb)
weights = list(divide(inv_corr_sum, inv_corr_sum.sum()))
- msgc = "max_div weight strategy failed: final weights contain NaN values"
+ msgc = "max_div weight strategy failed: final weights contain NaN values"
if any(isnan(weight) for weight in weights): # pragma: no cover
raise MaxDiversificationNaNError(msgc)
msgd = (
- "max_div weight strategy failed: negative weights detected"
- f" - weights: {[round(w, 6) for w in weights]}"
+ "max_div weight strategy failed: negative weights detected"
+ f" - weights: {[round(w, 6) for w in weights]}"
)
if any(weight < 0 for weight in weights):
raise MaxDiversificationNegativeWeightsError(msgd)
except linalg.LinAlgError as e:
msge = (
- "max_div weight strategy failed: "
- f"correlation matrix is singular - {e!s}"
+ "max_div weight strategy failed: "
+ f"correlation matrix is singular - {e!s}"
)
raise MaxDiversificationNaNError(msge) from e
else:
@@ -1495,14 +1495,14 @@ Source code for openseries.frame
self: Self,
returns: DataFrame,
) -> list[float]:
- """Calculate minimum volatility overweight weights.
+ """Calculate minimum volatility overweight weights.
Args:
returns: Returns DataFrame.
Returns:
List of minimum volatility overweight weights.
- """
+ """
vols = std(returns, axis=0, ddof=1)
min_vol_idx = vols.argmin()
min_vol_weight = 0.6
@@ -1516,7 +1516,7 @@ Source code for openseries.frame
weight_strat: LiteralPortfolioWeightings,
returns: DataFrame,
) -> list[float]:
- """Calculate weights based on strategy.
+ """Calculate weights based on strategy.
Args:
weight_strat: Weight calculation strategy.
@@ -1527,17 +1527,17 @@ Source code for openseries.frame
Raises:
NotImplementedError: If strategy is not implemented.
- """
- if weight_strat == "eq_weights":
+ """
+ if weight_strat == "eq_weights":
return self._calculate_eq_weights()
- if weight_strat == "inv_vol":
+ if weight_strat == "inv_vol":
return self._calculate_inv_vol_weights(returns)
- if weight_strat == "max_div":
+ if weight_strat == "max_div":
return self._calculate_max_div_weights(returns)
- if weight_strat == "min_vol_overweight":
+ if weight_strat == "min_vol_overweight":
return self._calculate_min_vol_overweight_weights(returns)
- msg = "Weight strategy not implemented"
+ msg = "Weight strategy not implemented"
raise NotImplementedError(msg)
@@ -1547,7 +1547,7 @@
Source code for openseries.frame
name: str,
weight_strat: LiteralPortfolioWeightings | None = None,
) -> DataFrame:
- """Calculate a basket timeseries based on the supplied weights.
+ """Calculate a basket timeseries based on the supplied weights.
Args:
name: Name of the basket timeseries.
@@ -1555,11 +1555,11 @@ Source code for openseries.frame
Returns:
A basket timeseries.
- """
+ """
if self.weights is None and weight_strat is None:
msg = (
- "OpenFrame weights property must be provided "
- "to run the make_portfolio method."
+ "OpenFrame weights property must be provided "
+ "to run the make_portfolio method."
)
raise NoWeightsError(msg)
@@ -1575,7 +1575,7 @@ Source code for openseries.frame
data=(returns @ array(self.weights)).add(1.0).cumprod(),
index=self.tsdf.index,
columns=[[name], [ValueType.PRICE]],
- dtype="float64",
+ dtype="float64",
)
@@ -1588,7 +1588,7 @@ Source code for openseries.frame
observations: int = 21,
periods_in_a_year_fixed: DaysInYearType | None = None,
) -> DataFrame:
- """Calculate rolling Information Ratio.
+ """Calculate rolling Information Ratio.
The Information Ratio equals ( fund return less index return ) divided by
the Tracking Error. And the Tracking Error is the standard deviation of the
@@ -1606,16 +1606,16 @@ Source code for openseries.frame
Returns:
Rolling Information Ratios.
- """
+ """
long_label = cast(
- "tuple[str, str]",
+ "tuple[str, str]",
self.tsdf.iloc[:, long_column].name,
)[0]
short_label = cast(
- "tuple[str, str]",
+ "tuple[str, str]",
self.tsdf.iloc[:, short_column].name,
)[0]
- ratio_label = f"{long_label} / {short_label}"
+ ratio_label = f"{long_label} / {short_label}"
if periods_in_a_year_fixed:
time_factor = float(periods_in_a_year_fixed)
else:
@@ -1640,7 +1640,7 @@ Source code for openseries.frame
voldf = voldf.dropna().to_frame()
ratiodf = (retdf.iloc[:, 0] / voldf.iloc[:, 0]).to_frame()
- ratiodf.columns = [[ratio_label], ["Information Ratio"]]
+ ratiodf.columns = [[ratio_label], ["Information Ratio"]]
return DataFrame(ratiodf)
@@ -1654,7 +1654,7 @@ Source code for openseries.frame
observations: int = 21,
dlta_degr_freedms: int = 1,
) -> DataFrame:
- """Calculate rolling Market Beta.
+ """Calculate rolling Market Beta.
Calculates Beta as Co-variance of asset & market divided by Variance
of the market.
@@ -1671,12 +1671,12 @@ Source code for openseries.frame
Returns:
Rolling Betas.
- """
- market_label = cast("tuple[str, str]", self.tsdf.iloc[:, market_column].name)[
+ """
+ market_label = cast("tuple[str, str]", self.tsdf.iloc[:, market_column].name)[
0
]
- asset_label = cast("tuple[str, str]", self.tsdf.iloc[:, asset_column].name)[0]
- beta_label = f"{asset_label} / {market_label}"
+ asset_label = cast("tuple[str, str]", self.tsdf.iloc[:, asset_column].name)[0]
+ beta_label = f"{asset_label} / {market_label}"
rolling = (
self.tsdf.ffill()
@@ -1702,7 +1702,7 @@ Source code for openseries.frame
)
rollbeta = rollbetaseries.to_frame()
rollbeta.index = rollbeta.index.get_level_values(0)
- rollbeta.columns = MultiIndex.from_arrays([[beta_label], ["Beta"]])
+ rollbeta.columns = MultiIndex.from_arrays([[beta_label], ["Beta"]])
return rollbeta
@@ -1715,7 +1715,7 @@ Source code for openseries.frame
second_column: int = 1,
observations: int = 21,
) -> DataFrame:
- """Calculate rolling Correlation.
+ """Calculate rolling Correlation.
Calculates correlation between two series. The period with
at least the given number of observations is the first period calculated.
@@ -1730,11 +1730,11 @@ Source code for openseries.frame
Returns:
Rolling Correlations.
- """
+ """
corr_label = (
- cast("tuple[str, str]", self.tsdf.iloc[:, first_column].name)[0]
- + "_VS_"
- + cast("tuple[str, str]", self.tsdf.iloc[:, second_column].name)[0]
+ cast("tuple[str, str]", self.tsdf.iloc[:, first_column].name)[0]
+ + "_VS_"
+ + cast("tuple[str, str]", self.tsdf.iloc[:, second_column].name)[0]
)
first_series = (
self.tsdf.iloc[:, first_column]
@@ -1747,7 +1747,7 @@ Source code for openseries.frame
corrdf.columns = MultiIndex.from_arrays(
[
[corr_label],
- ["Rolling correlation"],
+ ["Rolling correlation"],
],
)
@@ -1760,7 +1760,7 @@ Source code for openseries.frame
self: Self,
dependent_column: tuple[str, ValueType],
) -> tuple[DataFrame, OpenTimeSeries]:
- """Perform a multi-factor linear regression.
+ """Perform a multi-factor linear regression.
This function treats one specified column in the DataFrame as the dependent
variable (y) and uses all remaining columns as independent variables (X).
@@ -1780,28 +1780,28 @@ Source code for openseries.frame
Raises:
KeyError: If the column tuple is not found in the OpenFrame.tsdf.columns.
ValueError: If not all series are returnseries (ValueType.RTRN).
- """
+ """
key_msg = (
- f"Tuple ({dependent_column[0]}, "
- f"{dependent_column[1].value}) not found in data."
+ f"Tuple ({dependent_column[0]}, "
+ f"{dependent_column[1].value}) not found in data."
)
if dependent_column not in self.tsdf.columns:
raise KeyError(key_msg)
- vtype_msg = "All series should be of ValueType.RTRN."
+ vtype_msg = "All series should be of ValueType.RTRN."
if not all(x == ValueType.RTRN for x in self.tsdf.columns.get_level_values(1)):
raise MixedValuetypesError(vtype_msg)
dependent = self.tsdf[dependent_column]
factors = self.tsdf.drop(columns=[dependent_column])
- indx = ["R-square", "Intercept", *factors.columns.get_level_values(0)]
+ indx = ["R-square", "Intercept", *factors.columns.get_level_values(0)]
model = LinearRegression()
model.fit(factors, dependent)
predictions = OpenTimeSeries.from_arrays(
- name=f"Predicted {dependent_column[0]}",
- dates=[date.strftime("%Y-%m-%d") for date in self.tsdf.index],
+ name=f"Predicted {dependent_column[0]}",
+ dates=[date.strftime("%Y-%m-%d") for date in self.tsdf.index],
values=list(model.predict(factors)),
valuetype=ValueType.RTRN,
)
@@ -1820,7 +1820,7 @@ Source code for openseries.frame
*,
equal_weights: bool,
) -> tuple[list[str], list[float]]:
- """Validate and prepare inputs for rebalanced portfolio.
+ """Validate and prepare inputs for rebalanced portfolio.
Args:
items: List of items to include. If None, uses all items.
@@ -1834,30 +1834,30 @@ Source code for openseries.frame
WeightsNotProvidedError: If weights are required but not provided.
TypeError: If items is not a list.
PortfolioItemsNotWithinFrameError: If items are invalid.
- """
+ """
if bal_weights is None and not equal_weights:
if self.weights is None:
- msg = "Weights must be provided."
+ msg = "Weights must be provided."
raise WeightsNotProvidedError(msg)
bal_weights = list(self.weights)
if items is None:
items = list(self.columns_lvl_zero)
else:
- msg = "Items must be passed as list."
+ msg = "Items must be passed as list."
if not isinstance(items, list):
raise TypeError(msg)
if not items:
- msg = "Items for portfolio must be within SeriesFrame items."
+ msg = "Items for portfolio must be within SeriesFrame items."
raise PortfolioItemsNotWithinFrameError(msg)
if not set(items) <= set(self.columns_lvl_zero):
- msg = "Items for portfolio must be within SeriesFrame items."
+ msg = "Items for portfolio must be within SeriesFrame items."
raise PortfolioItemsNotWithinFrameError(msg)
if equal_weights:
bal_weights = [1 / len(items)] * len(items)
- return items, cast("list[float]", bal_weights)
+ return items, cast("list[float]", bal_weights)
def _initialize_rebalance_output(
self: Self,
@@ -1865,7 +1865,7 @@ Source code for openseries.frame
name: str,
cash_values: list[float],
) -> dict[str, dict[str, list[float]]]:
- """Initialize output structure for rebalanced portfolio.
+ """Initialize output structure for rebalanced portfolio.
Args:
items: List of items in portfolio.
@@ -1874,35 +1874,35 @@ Source code for openseries.frame
Returns:
Initialized output dictionary.
- """
+ """
output = {
item: {
ValueType.PRICE: [],
- "buysell_qty": [0.0] * self.length,
- "position": [0.0] * self.length,
- "value": [0.0] * self.length,
- "twr": [0.0] * self.length,
- "settle": [0.0] * self.length,
+ "buysell_qty": [0.0] * self.length,
+ "position": [0.0] * self.length,
+ "value": [0.0] * self.length,
+ "twr": [0.0] * self.length,
+ "settle": [0.0] * self.length,
}
for item in items
}
output.update(
{
- "cash": {
+ "cash": {
ValueType.PRICE: cash_values,
- "buysell_qty": [0.0] * self.length,
- "position": [0.0] * self.length,
- "value": [0.0] * self.length,
- "twr": [0.0] * self.length,
- "settle": [0.0] * self.length,
+ "buysell_qty": [0.0] * self.length,
+ "position": [0.0] * self.length,
+ "value": [0.0] * self.length,
+ "twr": [0.0] * self.length,
+ "settle": [0.0] * self.length,
},
name: {
ValueType.PRICE: [1.0] + [0.0] * (self.length - 1),
- "buysell_qty": [-1.0] + [0.0] * (self.length - 1),
- "position": [-1.0] + [0.0] * (self.length - 1),
- "value": [-1.0] + [0.0] * (self.length - 1),
- "twr": [1.0] + [0.0] * (self.length - 1),
- "settle": [1.0] + [0.0] * (self.length - 1),
+ "buysell_qty": [-1.0] + [0.0] * (self.length - 1),
+ "position": [-1.0] + [0.0] * (self.length - 1),
+ "value": [-1.0] + [0.0] * (self.length - 1),
+ "twr": [1.0] + [0.0] * (self.length - 1),
+ "settle": [1.0] + [0.0] * (self.length - 1),
},
},
)
@@ -1915,38 +1915,38 @@ Source code for openseries.frame
output: dict[str, dict[str, list[float]]],
name: str,
) -> None:
- """Initialize positions for the first day.
+ """Initialize positions for the first day.
Args:
items: List of items in portfolio.
bal_weights: Weights for each item.
output: Output dictionary to update.
name: Name of the portfolio.
- """
+ """
for item, weight in zip(items, bal_weights, strict=False):
output[item][ValueType.PRICE] = cast(
- "list[float]",
+ "list[float]",
self.tsdf[(item, ValueType.PRICE)].to_numpy().tolist(),
)
- output[item]["buysell_qty"][0] = (
+ output[item]["buysell_qty"][0] = (
weight / self.tsdf[(item, ValueType.PRICE)].iloc[0]
)
- output[item]["position"][0] = output[item]["buysell_qty"][0]
- output[item]["value"][0] = (
- output[item]["position"][0] * output[item][ValueType.PRICE][0]
+ output[item]["position"][0] = output[item]["buysell_qty"][0]
+ output[item]["value"][0] = (
+ output[item]["position"][0] * output[item][ValueType.PRICE][0]
)
- output[item]["settle"][0] = (
- -output[item]["buysell_qty"][0] * output[item][ValueType.PRICE][0]
+ output[item]["settle"][0] = (
+ -output[item]["buysell_qty"][0] * output[item][ValueType.PRICE][0]
)
- output["cash"]["buysell_qty"][0] += output[item]["settle"][0]
- output[item]["twr"][0] = (
- output[item]["value"][0] / -output[item]["settle"][0]
+ output["cash"]["buysell_qty"][0] += output[item]["settle"][0]
+ output[item]["twr"][0] = (
+ output[item]["value"][0] / -output[item]["settle"][0]
)
- output["cash"]["position"][0] = (
- output["cash"]["buysell_qty"][0] + output[name]["settle"][0]
+ output["cash"]["position"][0] = (
+ output["cash"]["buysell_qty"][0] + output[name]["settle"][0]
)
- output["cash"]["settle"][0] = -output["cash"]["position"][0]
+ output["cash"]["settle"][0] = -output["cash"]["position"][0]
def _process_rebalancing_day(
self: Self,
@@ -1956,7 +1956,7 @@ Source code for openseries.frame
output: dict[str, dict[str, list[float]]],
name: str,
) -> tuple[float, float]:
- """Process a rebalancing day.
+ """Process a rebalancing day.
Args:
day: Current day index.
@@ -1967,31 +1967,31 @@ Source code for openseries.frame
Returns:
Tuple of (portfolio_value, settle_value).
- """
+ """
portfolio_value = 0.0
settle_value = 0.0
for item, weight in zip(items, bal_weights, strict=False):
- output[item]["buysell_qty"][day] = (
+ output[item]["buysell_qty"][day] = (
weight
- - output[item]["value"][day - 1] / -output[name]["value"][day - 1]
+ - output[item]["value"][day - 1] / -output[name]["value"][day - 1]
) / output[item][ValueType.PRICE][day]
- output[item]["position"][day] = (
- output[item]["position"][day - 1] + output[item]["buysell_qty"][day]
+ output[item]["position"][day] = (
+ output[item]["position"][day - 1] + output[item]["buysell_qty"][day]
)
- output[item]["value"][day] = (
- output[item]["position"][day] * output[item][ValueType.PRICE][day]
+ output[item]["value"][day] = (
+ output[item]["position"][day] * output[item][ValueType.PRICE][day]
)
- portfolio_value += output[item]["value"][day]
- output[item]["twr"][day] = (
- output[item]["value"][day]
- / (output[item]["value"][day - 1] - output[item]["settle"][day])
- * output[item]["twr"][day - 1]
+ portfolio_value += output[item]["value"][day]
+ output[item]["twr"][day] = (
+ output[item]["value"][day]
+ / (output[item]["value"][day - 1] - output[item]["settle"][day])
+ * output[item]["twr"][day - 1]
)
- output[item]["settle"][day] = (
- -output[item]["buysell_qty"][day] * output[item][ValueType.PRICE][day]
+ output[item]["settle"][day] = (
+ -output[item]["buysell_qty"][day] * output[item][ValueType.PRICE][day]
)
- settle_value += output[item]["settle"][day]
+ settle_value += output[item]["settle"][day]
return portfolio_value, settle_value
@@ -2001,7 +2001,7 @@ Source code for openseries.frame
items: list[str],
output: dict[str, dict[str, list[float]]],
) -> float:
- """Process a non-rebalancing day.
+ """Process a non-rebalancing day.
Args:
day: Current day index.
@@ -2010,19 +2010,19 @@ Source code for openseries.frame
Returns:
Portfolio value.
- """
+ """
portfolio_value = 0.0
for item in items:
- output[item]["position"][day] = output[item]["position"][day - 1]
- output[item]["value"][day] = (
- output[item]["position"][day] * output[item][ValueType.PRICE][day]
+ output[item]["position"][day] = output[item]["position"][day - 1]
+ output[item]["value"][day] = (
+ output[item]["position"][day] * output[item][ValueType.PRICE][day]
)
- portfolio_value += output[item]["value"][day]
- output[item]["twr"][day] = (
- output[item]["value"][day]
- / (output[item]["value"][day - 1] - output[item]["settle"][day])
- * output[item]["twr"][day - 1]
+ portfolio_value += output[item]["value"][day]
+ output[item]["twr"][day] = (
+ output[item]["value"][day]
+ / (output[item]["value"][day - 1] - output[item]["settle"][day])
+ * output[item]["twr"][day - 1]
)
return portfolio_value
@@ -2035,7 +2035,7 @@ Source code for openseries.frame
output: dict[str, dict[str, list[float]]],
name: str,
) -> None:
- """Update cash and portfolio values for a day.
+ """Update cash and portfolio values for a day.
Args:
day: Current day index.
@@ -2043,22 +2043,22 @@ Source code for openseries.frame
settle_value: Total settle value.
output: Output dictionary to update.
name: Name of the portfolio.
- """
- output["cash"]["buysell_qty"][day] = settle_value
- output["cash"]["position"][day] = (
- output["cash"]["position"][day - 1]
- * output["cash"][ValueType.PRICE][day]
- / output["cash"][ValueType.PRICE][day - 1]
- + output["cash"]["buysell_qty"][day]
+ """
+ output["cash"]["buysell_qty"][day] = settle_value
+ output["cash"]["position"][day] = (
+ output["cash"]["position"][day - 1]
+ * output["cash"][ValueType.PRICE][day]
+ / output["cash"][ValueType.PRICE][day - 1]
+ + output["cash"]["buysell_qty"][day]
)
- output["cash"]["value"][day] = output["cash"]["position"][day]
- total_portfolio_value = portfolio_value + output["cash"]["value"][day]
- output[name]["position"][day] = output[name]["position"][day - 1]
- output[name]["value"][day] = -total_portfolio_value
- output[name]["twr"][day] = (
- output[name]["value"][day] / output[name]["position"][day]
+ output["cash"]["value"][day] = output["cash"]["position"][day]
+ total_portfolio_value = portfolio_value + output["cash"]["value"][day]
+ output[name]["position"][day] = output[name]["position"][day - 1]
+ output[name]["value"][day] = -total_portfolio_value
+ output[name]["twr"][day] = (
+ output[name]["value"][day] / output[name]["position"][day]
)
- output[name][ValueType.PRICE][day] = output[name]["twr"][day]
+ output[name][ValueType.PRICE][day] = output[name]["twr"][day]
def _build_rebalance_result(
self: Self,
@@ -2066,7 +2066,7 @@ Source code for openseries.frame
instruments: list[str],
subheaders: list[str | ValueType],
) -> DataFrame:
- """Build result DataFrame from output dictionary.
+ """Build result DataFrame from output dictionary.
Args:
output: Output dictionary with all calculated values.
@@ -2075,7 +2075,7 @@ Source code for openseries.frame
Returns:
DataFrame with MultiIndex columns.
- """
+ """
result = DataFrame()
for outvalue in output.values():
result = concat(
@@ -2083,7 +2083,7 @@ Source code for openseries.frame
result,
DataFrame(data=outvalue, index=self.tsdf.index),
],
- axis="columns",
+ axis="columns",
)
lvlone, lvltwo = [], []
for instr in instruments:
@@ -2105,7 +2105,7 @@ Source code for openseries.frame
equal_weights: bool = False,
drop_extras: bool = True,
) -> OpenFrame:
- """Create a rebalanced portfolio from the OpenFrame constituents.
+ """Create a rebalanced portfolio from the OpenFrame constituents.
Args:
name: Name of the portfolio.
@@ -2121,7 +2121,7 @@ Source code for openseries.frame
Returns:
OpenFrame containing the rebalanced portfolio.
- """
+ """
items, bal_weights = self._validate_and_prepare_rebalance_inputs(
items,
bal_weights,
@@ -2131,7 +2131,7 @@ Source code for openseries.frame
if cash_index:
cash_index.tsdf = cash_index.tsdf.reindex(self.tsdf.index)
cash_values: list[float] = cast(
- "list[float]", cash_index.tsdf.iloc[:, 0].to_numpy().tolist()
+ "list[float]", cash_index.tsdf.iloc[:, 0].to_numpy().tolist()
)
else:
cash_values = [1.0] * self.length
@@ -2141,18 +2141,18 @@ Source code for openseries.frame
ccies = list({serie.currency for serie in self.constituents})
if len(ccies) != 1:
- msg = "Items for portfolio must be denominated in same currency."
+ msg = "Items for portfolio must be denominated in same currency."
raise MultipleCurrenciesError(msg)
currency = ccies[0]
- instruments = [*items, "cash", name]
+ instruments = [*items, "cash", name]
subheaders = [
ValueType.PRICE,
- "buysell_qty",
- "position",
- "value",
- "twr",
- "settle",
+ "buysell_qty",
+ "position",
+ "value",
+ "twr",
+ "settle",
]
output = self._initialize_rebalance_output(
@@ -2209,7 +2209,7 @@ Source code for openseries.frame
series.extend(
[
OpenTimeSeries.from_df(
- dframe=result[(item.label, "twr")],
+ dframe=result[(item.label, "twr")],
valuetype=ValueType.PRICE,
baseccy=item.currency,
local_ccy=item.local_ccy,
@@ -2219,7 +2219,7 @@ Source code for openseries.frame
)
series.append(
OpenTimeSeries.from_df(
- dframe=result[(name, "twr")],
+ dframe=result[(name, "twr")],
valuetype=ValueType.PRICE,
baseccy=currency,
local_ccy=True,
@@ -2233,7 +2233,7 @@ Source code for openseries.frame
valuetype=ValueType.PRICE,
baseccy=currency,
local_ccy=True,
- ).set_new_label(f"{col[0]}, {col[1]!s}")
+ ).set_new_label(f"{col[0]}, {col[1]!s}")
for col in result.columns
]
)
diff --git a/docs/build/html/_modules/openseries/load_plotly.html b/docs/build/html/_modules/openseries/load_plotly.html
index 0b948bf6..ebab9690 100644
--- a/docs/build/html/_modules/openseries/load_plotly.html
+++ b/docs/build/html/_modules/openseries/load_plotly.html
@@ -108,7 +108,7 @@
Source code for openseries.load_plotly
-"""Function to load plotly layout and configuration from local json file."""
+"""Function to load plotly layout and configuration from local json file."""
from __future__ import annotations
@@ -125,18 +125,18 @@ Source code for openseries.load_plotly
logger = getLogger(__name__)
-__all__ = ["load_plotly_dict"]
+__all__ = ["load_plotly_dict"]
def _check_remote_file_existence(url: str) -> bool:
- """Check if remote file exists.
+ """Check if remote file exists.
Args:
url: Path to remote file.
Returns:
True if url is valid and False otherwise.
- """
+ """
ok_code = 200
try:
@@ -154,7 +154,7 @@ Source code for openseries.load_plotly
*,
responsive: bool = True,
) -> tuple[PlotlyLayoutType, CaptorLogoType]:
- """Load Plotly defaults.
+ """Load Plotly defaults.
Args:
responsive: Flag whether to load as responsive. Defaults to True.
@@ -164,22 +164,22 @@ Source code for openseries.load_plotly
where config_and_layout is the Plotly config and layout template dict,
and logo is the Captor logo dict (may be empty if the remote logo is
unavailable).
- """
+ """
package_dir = Path(__file__).parent
- layoutfile = package_dir / "plotly_layouts.json"
- logofile = package_dir / "plotly_captor_logo.json"
+ layoutfile = package_dir / "plotly_layouts.json"
+ logofile = package_dir / "plotly_captor_logo.json"
- with layoutfile.open(mode="r", encoding="utf-8") as layout_file:
+ with layoutfile.open(mode="r", encoding="utf-8") as layout_file:
fig = load(layout_file)
- with logofile.open(mode="r", encoding="utf-8") as logo_file:
+ with logofile.open(mode="r", encoding="utf-8") as logo_file:
logo = load(logo_file)
- if not _check_remote_file_existence(url=logo["source"]):
- msg = f"Failed to add logo image from URL {logo['source']}"
+ if not _check_remote_file_existence(url=logo["source"]):
+ msg = f"Failed to add logo image from URL {logo['source']}"
logger.warning(msg)
logo = {}
- fig["config"].update({"responsive": responsive})
+ fig["config"].update({"responsive": responsive})
return fig, logo
diff --git a/docs/build/html/_modules/openseries/owntypes.html b/docs/build/html/_modules/openseries/owntypes.html
index b2b0cc33..592963dc 100644
--- a/docs/build/html/_modules/openseries/owntypes.html
+++ b/docs/build/html/_modules/openseries/owntypes.html
@@ -108,7 +108,7 @@
Source code for openseries.owntypes
-"""Declaring types used throughout the project."""
+"""Declaring types used throughout the project."""
from __future__ import annotations
@@ -137,17 +137,17 @@ Source code for openseries.owntypes
else:
SeriesFloat = Series
-__all__ = ["ValueType"]
+__all__ = ["ValueType"]
-SeriesOrFloat_co = TypeVar("SeriesOrFloat_co", float, SeriesFloat, covariant=True)
+SeriesOrFloat_co = TypeVar("SeriesOrFloat_co", float, SeriesFloat, covariant=True)
CountryStringType = Annotated[
str,
StringConstraints(
strip_whitespace=True,
- pattern=r"^[A-Z]{2}$",
+ pattern=r"^[A-Z]{2}$",
to_upper=True,
min_length=2,
max_length=2,
@@ -161,7 +161,7 @@ Source code for openseries.owntypes
[docs]
class Countries(BaseModel):
-
"""Declare Countries."""
+
"""Declare Countries."""
countryinput: CountriesType
@@ -170,7 +170,7 @@ Source code for openseries.owntypes
CurrencyStringType = Annotated[
str,
StringConstraints(
- pattern=r"^[A-Z]{3}$",
+ pattern=r"^[A-Z]{3}$",
to_upper=True,
min_length=3,
max_length=3,
@@ -183,7 +183,7 @@ Source code for openseries.owntypes
[docs]
class Currency(BaseModel):
-
"""Declare Currency."""
+
"""Declare Currency."""
ccy: CurrencyStringType
@@ -192,7 +192,7 @@ Source code for openseries.owntypes
DateStringType = Annotated[
str,
StringConstraints(
- pattern=r"^\d{4}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$",
+ pattern=r"^\d{4}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$",
strip_whitespace=True,
strict=True,
min_length=10,
@@ -220,153 +220,153 @@ Source code for openseries.owntypes
CaptorLogoType = dict[str, str | float]
-LiteralJsonOutput = Literal["values", "tsdf"]
-LiteralTrunc = Literal["before", "after", "both"]
+LiteralJsonOutput = Literal["values", "tsdf"]
+LiteralTrunc = Literal["before", "after", "both"]
LiteralLinePlotMode = (
Literal[
- "lines",
- "markers",
- "lines+markers",
- "lines+text",
- "markers+text",
- "lines+markers+text",
+ "lines",
+ "markers",
+ "lines+markers",
+ "lines+text",
+ "markers+text",
+ "lines+markers+text",
]
| None
)
-LiteralHowMerge = Literal["outer", "inner"]
-LiteralQuantileInterp = Literal["linear", "lower", "higher", "midpoint", "nearest"]
-LiteralBizDayFreq = Literal["B", "BME", "BQE", "BYE"]
+LiteralHowMerge = Literal["outer", "inner"]
+LiteralQuantileInterp = Literal["linear", "lower", "higher", "midpoint", "nearest"]
+LiteralBizDayFreq = Literal["B", "BME", "BQE", "BYE"]
LiteralPandasReindexMethod = (
- Literal["pad", "ffill", "backfill", "bfill", "nearest"] | None
+ Literal["pad", "ffill", "backfill", "bfill", "nearest"] | None
)
-LiteralNanMethod = Literal["fill", "drop"]
-LiteralCaptureRatio = Literal["up", "down", "both"]
-LiteralBarPlotMode = Literal["stack", "group", "overlay", "relative"]
-LiteralPlotlyOutput = Literal["file", "div"]
-LiteralPlotlyJSlib = Literal[True, False, "cdn"]
-LiteralPlotlyHistogramPlotType = Literal["bars", "lines"]
-LiteralPlotlyHistogramBarMode = Literal["stack", "group", "overlay", "relative"]
-LiteralPlotlyHistogramCurveType = Literal["normal", "kde"]
+LiteralNanMethod = Literal["fill", "drop"]
+LiteralCaptureRatio = Literal["up", "down", "both"]
+LiteralBarPlotMode = Literal["stack", "group", "overlay", "relative"]
+LiteralPlotlyOutput = Literal["file", "div"]
+LiteralPlotlyJSlib = Literal[True, False, "cdn"]
+LiteralPlotlyHistogramPlotType = Literal["bars", "lines"]
+LiteralPlotlyHistogramBarMode = Literal["stack", "group", "overlay", "relative"]
+LiteralPlotlyHistogramCurveType = Literal["normal", "kde"]
LiteralPlotlyHistogramHistNorm = Literal[
- "percent",
- "probability",
- "density",
- "probability density",
+ "percent",
+ "probability",
+ "density",
+ "probability density",
]
LiteralPortfolioWeightings = Literal[
- "eq_weights", "inv_vol", "max_div", "min_vol_overweight"
+ "eq_weights", "inv_vol", "max_div", "min_vol_overweight"
]
LiteralMinimizeMethods = Literal[
- "SLSQP",
- "Nelder-Mead",
- "Powell",
- "CG",
- "BFGS",
- "Newton-CG",
- "L-BFGS-B",
- "TNC",
- "COBYLA",
- "trust-constr",
- "dogleg",
- "trust-ncg",
- "trust-exact",
- "trust-krylov",
+ "SLSQP",
+ "Nelder-Mead",
+ "Powell",
+ "CG",
+ "BFGS",
+ "Newton-CG",
+ "L-BFGS-B",
+ "TNC",
+ "COBYLA",
+ "trust-constr",
+ "dogleg",
+ "trust-ncg",
+ "trust-exact",
+ "trust-krylov",
]
LiteralSeriesProps = Literal[
- "value_ret",
- "geo_ret",
- "arithmetic_ret",
- "vol",
- "downside_deviation",
- "ret_vol_ratio",
- "sortino_ratio",
- "kappa3_ratio",
- "z_score",
- "skew",
- "kurtosis",
- "positive_share",
- "var_down",
- "cvar_down",
- "vol_from_var",
- "worst",
- "worst_month",
- "max_drawdown_cal_year",
- "max_drawdown",
- "max_drawdown_date",
- "first_idx",
- "last_idx",
- "length",
- "span_of_days",
- "yearfrac",
- "periods_in_a_year",
- "autocorr",
- "partial_autocorr",
+ "value_ret",
+ "geo_ret",
+ "arithmetic_ret",
+ "vol",
+ "downside_deviation",
+ "ret_vol_ratio",
+ "sortino_ratio",
+ "kappa3_ratio",
+ "z_score",
+ "skew",
+ "kurtosis",
+ "positive_share",
+ "var_down",
+ "cvar_down",
+ "vol_from_var",
+ "worst",
+ "worst_month",
+ "max_drawdown_cal_year",
+ "max_drawdown",
+ "max_drawdown_date",
+ "first_idx",
+ "last_idx",
+ "length",
+ "span_of_days",
+ "yearfrac",
+ "periods_in_a_year",
+ "autocorr",
+ "partial_autocorr",
]
LiteralFrameProps = Literal[
- "value_ret",
- "geo_ret",
- "arithmetic_ret",
- "autocorr",
- "vol",
- "downside_deviation",
- "ret_vol_ratio",
- "sortino_ratio",
- "kappa3_ratio",
- "z_score",
- "skew",
- "kurtosis",
- "positive_share",
- "var_down",
- "cvar_down",
- "vol_from_var",
- "worst",
- "worst_month",
- "max_drawdown",
- "max_drawdown_date",
- "max_drawdown_cal_year",
- "first_indices",
- "last_indices",
- "lengths_of_items",
- "span_of_days_all",
+ "value_ret",
+ "geo_ret",
+ "arithmetic_ret",
+ "autocorr",
+ "vol",
+ "downside_deviation",
+ "ret_vol_ratio",
+ "sortino_ratio",
+ "kappa3_ratio",
+ "z_score",
+ "skew",
+ "kurtosis",
+ "positive_share",
+ "var_down",
+ "cvar_down",
+ "vol_from_var",
+ "worst",
+ "worst_month",
+ "max_drawdown",
+ "max_drawdown_date",
+ "max_drawdown_cal_year",
+ "first_indices",
+ "last_indices",
+ "lengths_of_items",
+ "span_of_days_all",
]
[docs]
class PropertiesList(list[str]):
-
"""Base class for allowed property arguments definition."""
+
"""Base class for allowed property arguments definition."""
allowed_strings: ClassVar[set[str]] = {
-
"value_ret",
-
"geo_ret",
-
"arithmetic_ret",
-
"vol",
-
"downside_deviation",
-
"ret_vol_ratio",
-
"sortino_ratio",
-
"kappa3_ratio",
-
"omega_ratio",
-
"z_score",
-
"skew",
-
"kurtosis",
-
"positive_share",
-
"var_down",
-
"cvar_down",
-
"vol_from_var",
-
"worst",
-
"worst_month",
-
"max_drawdown",
-
"max_drawdown_date",
-
"max_drawdown_cal_year",
+
"value_ret",
+
"geo_ret",
+
"arithmetic_ret",
+
"vol",
+
"downside_deviation",
+
"ret_vol_ratio",
+
"sortino_ratio",
+
"kappa3_ratio",
+
"omega_ratio",
+
"z_score",
+
"skew",
+
"kurtosis",
+
"positive_share",
+
"var_down",
+
"cvar_down",
+
"vol_from_var",
+
"worst",
+
"worst_month",
+
"max_drawdown",
+
"max_drawdown_date",
+
"max_drawdown_cal_year",
}
def _validate(self: Self) -> None:
-
"""Validate the string input of the all_properties method."""
+
"""Validate the string input of the all_properties method."""
seen = set()
invalids = set()
duplicates = set()
-
msg = ""
+
msg = ""
for item in self:
if item not in self.allowed_strings:
invalids.add(item)
@@ -375,11 +375,11 @@
Source code for openseries.owntypes
seen.add(item)
if len(invalids) != 0:
msg += (
- f"Invalid string(s): {list(invalids)}.\nAllowed strings are:"
- f"\n{pformat(self.allowed_strings)}\n"
+ f"Invalid string(s): {list(invalids)}.\nAllowed strings are:"
+ f"\n{pformat(self.allowed_strings)}\n"
)
if len(duplicates) != 0:
- msg += f"Duplicate string(s): {list(duplicates)}."
+ msg += f"Duplicate string(s): {list(duplicates)}."
if len(msg) != 0:
raise PropertiesInputValidationError(msg)
@@ -388,17 +388,17 @@
Source code for openseries.owntypes
[docs]
class OpenTimeSeriesPropertiesList(PropertiesList):
-
"""Allowed property arguments for the OpenTimeSeries class."""
+
"""Allowed property arguments for the OpenTimeSeries class."""
allowed_strings: ClassVar[set[str]] = PropertiesList.allowed_strings | {
-
"first_idx",
-
"last_idx",
-
"length",
-
"span_of_days",
-
"yearfrac",
-
"periods_in_a_year",
-
"autocorr",
-
"partial_autocorr",
+
"first_idx",
+
"last_idx",
+
"length",
+
"span_of_days",
+
"yearfrac",
+
"periods_in_a_year",
+
"autocorr",
+
"partial_autocorr",
}
@@ -407,7 +407,7 @@
Source code for openseries.owntypes
self: Self,
*args: LiteralSeriesProps,
) -> None:
- """Property arguments for the OpenTimeSeries class."""
+ """Property arguments for the OpenTimeSeries class."""
super().__init__(args)
self._validate()
@@ -417,20 +417,20 @@
Source code for openseries.owntypes
[docs]
class OpenFramePropertiesList(PropertiesList):
-
"""Allowed property arguments for the OpenFrame class."""
+
"""Allowed property arguments for the OpenFrame class."""
allowed_strings: ClassVar[set[str]] = PropertiesList.allowed_strings | {
-
"autocorr",
-
"first_indices",
-
"last_indices",
-
"lengths_of_items",
-
"span_of_days_all",
+
"autocorr",
+
"first_indices",
+
"last_indices",
+
"lengths_of_items",
+
"span_of_days_all",
}
[docs]
def __init__(self: Self, *args: LiteralFrameProps) -> None:
-
"""Property arguments for the OpenFrame class."""
+
"""Property arguments for the OpenFrame class."""
super().__init__(args)
self._validate()
@@ -440,153 +440,153 @@ Source code for openseries.owntypes
[docs]
class ValueType(StrEnum):
-
"""Enum types of OpenTimeSeries to identify the output."""
+
"""Enum types of OpenTimeSeries to identify the output."""
-
EWMA_VOL = "EWMA volatility"
-
EWMA_VAR = "EWMA VaR"
-
PRICE = "Price(Close)"
-
RTRN = "Return(Total)"
-
RELRTRN = "Relative return"
-
ROLLBETA = "Beta"
-
ROLLCORR = "Rolling correlation"
-
ROLLCVAR = "Rolling CVaR"
-
ROLLINFORATIO = "Information Ratio"
-
ROLLRTRN = "Rolling returns"
-
ROLLVAR = "Rolling VaR"
-
ROLLVOL = "Rolling volatility"
+ EWMA_VOL = "EWMA volatility"
+ EWMA_VAR = "EWMA VaR"
+ PRICE = "Price(Close)"
+ RTRN = "Return(Total)"
+ RELRTRN = "Relative return"
+ ROLLBETA = "Beta"
+ ROLLCORR = "Rolling correlation"
+ ROLLCVAR = "Rolling CVaR"
+ ROLLINFORATIO = "Information Ratio"
+ ROLLRTRN = "Rolling returns"
+ ROLLVAR = "Rolling VaR"
+ ROLLVOL = "Rolling volatility"
[docs]
class MixedValuetypesError(Exception):
-
"""Raised when provided timeseries valuetypes are not the same."""
+ """Raised when provided timeseries valuetypes are not the same."""
[docs]
class AtLeastOneFrameError(Exception):
-
"""Raised when none of the possible frame inputs is provided."""
+
"""Raised when none of the possible frame inputs is provided."""
[docs]
class DateAlignmentError(Exception):
-
"""Raised when date input is not aligned with existing range."""
+ """Raised when date input is not aligned with existing range."""
[docs]
class NumberOfItemsAndLabelsNotSameError(Exception):
-
"""Raised when number of labels is not matching the number of timeseries."""
+
"""Raised when number of labels is not matching the number of timeseries."""
[docs]
class InitialValueZeroError(Exception):
-
"""Raised when a calculation cannot be performed due to initial value(s) zero."""
+ """Raised when a calculation cannot be performed due to initial value(s) zero."""
[docs]
class CountriesNotStringNorListStrError(Exception):
-
"""Raised when countries argument is not provided in correct format."""
+ """Raised when countries argument is not provided in correct format."""
[docs]
class MarketsNotStringNorListStrError(Exception):
-
"""Raised when markets argument is not provided in correct format."""
+ """Raised when markets argument is not provided in correct format."""
[docs]
class TradingDaysNotAboveZeroError(Exception):
-
"""Raised when trading days argument is not above zero."""
+ """Raised when trading days argument is not above zero."""
[docs]
class BothStartAndEndError(Exception):
-
"""Raised when both start and end dates are provided."""
+ """Raised when both start and end dates are provided."""
[docs]
class NoWeightsError(Exception):
-
"""Raised when no weights are provided to function where necessary."""
+ """Raised when no weights are provided to function where necessary."""
[docs]
class LabelsNotUniqueError(Exception):
-
"""Raised when provided label names are not unique."""
+ """Raised when provided label names are not unique."""
+
"""Raised when ratio keyword not provided correctly."""
[docs]
class MergingResultedInEmptyError(Exception):
-
"""Raised when a merge resulted in an empty DataFrame."""
+ """Raised when a merge resulted in an empty DataFrame."""
[docs]
class IncorrectArgumentComboError(Exception):
-
"""Raised when correct combination of arguments is not provided."""
+ """Raised when correct combination of arguments is not provided."""
+ """Raised when duplicate strings are provided."""
[docs]
class ResampleDataLossError(Exception):
-
"""Raised when user attempts to run resample_to_business_period_ends on returns."""
+
"""Raised when user attempts to run resample_to_business_period_ends on returns."""
class WeightsNotProvidedError(Exception):
- """Raised when weights are not provided."""
+ """Raised when weights are not provided."""
class MultipleCurrenciesError(Exception):
- """Raised when multiple currencies are provided."""
+ """Raised when multiple currencies are provided."""
class PortfolioItemsNotWithinFrameError(Exception):
- """Raised when portfolio items are not within frame."""
+ """Raised when portfolio items are not within frame."""
class MaxDiversificationNaNError(Exception):
- """Raised when max_div weight strategy produces NaN values."""
+ """Raised when max_div weight strategy produces NaN values."""
class MaxDiversificationNegativeWeightsError(Exception):
- """Raised when max_div weight strategy produces negative weights."""
+ """Raised when max_div weight strategy produces negative weights."""
diff --git a/docs/build/html/_modules/openseries/portfoliotools.html b/docs/build/html/_modules/openseries/portfoliotools.html
index 35896198..7ef4f58a 100644
--- a/docs/build/html/_modules/openseries/portfoliotools.html
+++ b/docs/build/html/_modules/openseries/portfoliotools.html
@@ -108,7 +108,7 @@
Source code for openseries.portfoliotools
-"""Defining the portfolio tools for the OpenFrame class."""
+"""Defining the portfolio tools for the OpenFrame class."""
from __future__ import annotations
@@ -162,11 +162,11 @@ Source code for openseries.portfoliotools
from .frame import OpenFrame
__all__ = [
- "constrain_optimized_portfolios",
- "efficient_frontier",
- "prepare_plot_data",
- "sharpeplot",
- "simulate_portfolios",
+ "constrain_optimized_portfolios",
+ "efficient_frontier",
+ "prepare_plot_data",
+ "sharpeplot",
+ "simulate_portfolios",
]
@@ -177,7 +177,7 @@ Source code for openseries.portfoliotools
num_ports: int,
seed: int,
) -> DataFrame:
- """Generate random weights for simulated portfolios.
+ """Generate random weights for simulated portfolios.
Args:
simframe: Return data for portfolio constituents.
@@ -186,7 +186,7 @@ Source code for openseries.portfoliotools
Returns:
The resulting data.
- """
+ """
copi = simframe.from_deepcopy()
vtypes = [x == ValueType.RTRN for x in copi.tsdf.columns.get_level_values(1)]
@@ -196,7 +196,7 @@ Source code for openseries.portfoliotools
elif all(vtypes):
log_ret = copi.tsdf.copy()
else:
- msg = "Mix of series types will give inconsistent results"
+ msg = "Mix of series types will give inconsistent results"
raise MixedValuetypesError(msg)
log_ret.columns = log_ret.columns.get_level_values(0)
@@ -209,15 +209,15 @@ Source code for openseries.portfoliotools
all_weights = all_weights / all_weights.sum(axis=1, keepdims=True)
ret_arr = all_weights @ mean_returns
- vol_arr = sqrt(einsum("ij,jk,ik->i", all_weights, cov_matrix, all_weights))
+ vol_arr = sqrt(einsum("ij,jk,ik->i", all_weights, cov_matrix, all_weights))
sharpe_arr = ret_arr / vol_arr
simdf = concat(
[
- DataFrame({"stdev": vol_arr, "ret": ret_arr, "sharpe": sharpe_arr}),
+ DataFrame({"stdev": vol_arr, "ret": ret_arr, "sharpe": sharpe_arr}),
DataFrame(all_weights, columns=simframe.columns_lvl_zero),
],
- axis="columns",
+ axis="columns",
)
simdf = simdf.replace([inf, -inf], nan)
return simdf.dropna()
@@ -225,7 +225,7 @@ Source code for openseries.portfoliotools
def _prepare_returns_for_frontier(eframe: OpenFrame) -> tuple[DataFrame, OpenFrame]:
- """Prepare returns DataFrame for frontier calculation.
+ """Prepare returns DataFrame for frontier calculation.
Args:
eframe: Portfolio data.
@@ -235,7 +235,7 @@ Source code for openseries.portfoliotools
Raises:
MixedValuetypesError: If series types are mixed.
- """
+ """
if eframe.weights is None:
eframe.weights = [1.0 / eframe.item_count] * eframe.item_count
@@ -248,7 +248,7 @@ Source code for openseries.portfoliotools
elif all(vtypes):
log_ret = copi.tsdf.copy()
else:
- msg = "Mix of series types will give inconsistent results"
+ msg = "Mix of series types will give inconsistent results"
raise MixedValuetypesError(msg)
log_ret.columns = log_ret.columns.get_level_values(0)
@@ -260,7 +260,7 @@ Source code for openseries.portfoliotools
log_ret: DataFrame,
periods_in_a_year: float,
) -> tuple[float, float]:
- """Calculate frontier return bounds.
+ """Calculate frontier return bounds.
Args:
simulated: Simulated portfolios DataFrame.
@@ -269,9 +269,9 @@ Source code for openseries.portfoliotools
Returns:
Tuple of (min_return, max_return).
- """
- min_stdev_idx = simulated["stdev"].idxmin()
- frontier_min = cast("float", simulated.loc[min_stdev_idx, "ret"])
+ """
+ min_stdev_idx = simulated["stdev"].idxmin()
+ frontier_min = cast("float", simulated.loc[min_stdev_idx, "ret"])
arithmetic_means = array(log_ret.mean() * periods_in_a_year)
cleaned_arithmetic_means = arithmetic_means[~isnan(arithmetic_means)]
@@ -290,7 +290,7 @@ Source code for openseries.portfoliotools
bounds: tuple[tuple[float, float], ...],
minimize_method: LiteralMinimizeMethods,
) -> tuple[list[float], list[NDArray[float64]]]:
- """Build frontier line points.
+ """Build frontier line points.
Args:
log_ret: Returns DataFrame.
@@ -304,10 +304,10 @@ Source code for openseries.portfoliotools
Returns:
Tuple of (frontier_x, frontier_weights).
- """
+ """
def _check_sum(weights: NDArray[float64]) -> float:
- return cast("float", npsum(weights) - 1)
+ return cast("float", npsum(weights) - 1)
def _get_ret_vol_sr(
lg_ret: DataFrame,
@@ -317,7 +317,7 @@ Source code for openseries.portfoliotools
ret = npsum(lg_ret.mean() * weights) * per_in_yr
volatility = sqrt(weights.T @ (lg_ret.cov() * per_in_yr @ weights))
sr = ret / volatility
- return cast("NDArray[float64]", array([ret, volatility, sr]))
+ return cast("NDArray[float64]", array([ret, volatility, sr]))
def _diff_return(
lg_ret: DataFrame,
@@ -326,7 +326,7 @@ Source code for openseries.portfoliotools
poss_return: float,
) -> float64:
return cast(
- "float64",
+ "float64",
_get_ret_vol_sr(lg_ret=lg_ret, weights=weights, per_in_yr=per_in_yr)[0]
- poss_return,
)
@@ -335,7 +335,7 @@ Source code for openseries.portfoliotools
weights: NDArray[float64],
) -> float64:
return cast(
- "float64",
+ "float64",
_get_ret_vol_sr(
lg_ret=log_ret,
weights=weights,
@@ -349,12 +349,12 @@ Source code for openseries.portfoliotools
for possible_return in frontier_y:
cons = cast(
- "Any",
+ "Any",
[
- {"type": "eq", "fun": _check_sum},
+ {"type": "eq", "fun": _check_sum},
{
- "type": "eq",
- "fun": lambda w, poss_return=possible_return: _diff_return(
+ "type": "eq",
+ "fun": lambda w, poss_return=possible_return: _diff_return(
lg_ret=log_ret,
weights=w,
per_in_yr=periods_in_a_year,
@@ -372,8 +372,8 @@ Source code for openseries.portfoliotools
constraints=cons,
)
- frontier_x.append(result["fun"])
- frontier_weights.append(result["x"])
+ frontier_x.append(result["fun"])
+ frontier_weights.append(result["x"])
return frontier_x, frontier_weights
@@ -384,7 +384,7 @@ Source code for openseries.portfoliotools
frontier_weights: list[NDArray[float64]],
columns_lvl_zero: list[str],
) -> DataFrame:
- """Build frontier DataFrame.
+ """Build frontier DataFrame.
Args:
frontier_x: Frontier volatility values.
@@ -394,25 +394,25 @@ Source code for openseries.portfoliotools
Returns:
Frontier DataFrame.
- """
+ """
line_df = concat(
[
DataFrame(data=frontier_weights, columns=columns_lvl_zero),
- DataFrame({"stdev": frontier_x, "ret": frontier_y}),
+ DataFrame({"stdev": frontier_x, "ret": frontier_y}),
],
- axis="columns",
+ axis="columns",
)
- line_df["sharpe"] = line_df.ret / line_df.stdev
+ line_df["sharpe"] = line_df.ret / line_df.stdev
limit_small = 0.0001
line_df = line_df.mask(line_df.abs() < limit_small, 0.0)
weight_cols = columns_lvl_zero
- weight_header = "<br><br>Weights:<br>"
- line_df["text"] = line_df[weight_cols].apply(
+ weight_header = "<br><br>Weights:<br>"
+ line_df["text"] = line_df[weight_cols].apply(
lambda row: (
weight_header
- + "<br>".join([f"{row[col]:.1%} {col}" for col in weight_cols])
+ + "<br>".join([f"{row[col]:.1%} {col}" for col in weight_cols])
),
axis=1,
)
@@ -421,18 +421,18 @@ Source code for openseries.portfoliotools
def _apply_tweak(line_df: DataFrame) -> DataFrame:
- """Apply tweak to frontier DataFrame.
+ """Apply tweak to frontier DataFrame.
Args:
line_df: Frontier DataFrame.
Returns:
Tweaked DataFrame.
- """
+ """
limit_tweak = 0.001
- line_df["stdev_diff"] = line_df.stdev.ffill().pct_change()
+ line_df["stdev_diff"] = line_df.stdev.ffill().pct_change()
line_df = line_df.loc[line_df.stdev_diff.abs() > limit_tweak]
- return line_df.drop(columns="stdev_diff")
+ return line_df.drop(columns="stdev_diff")
def _create_optimization_functions(
@@ -443,7 +443,7 @@ Source code for openseries.portfoliotools