feat: add set_values_monthly() to macro

Author: chilango74

docs/matplotlib_ext/docscrape.py

@@ -397,7 +397,7 @@ def _parse(self):
             msg = "Docstring contains a Receives section but not Yields."
             raise ValueError(msg)
 
-        for (section, content) in sections:
+        for section, content in sections:
             if not section.startswith(".."):
                 section = (s.capitalize() for s in section.split(" "))
                 section = " ".join(section)
@@ -605,7 +605,6 @@ def __init__(self, obj, doc=None, config={}):
 
 
 class ClassDoc(NumpyDocString):
-
     extra_public_methods = ["__call__"]
 
     def __init__(self, cls, doc=None, modulename="", func_doc=FunctionDoc, config={}):

examples/02 index funds perfomance.ipynb

@@ -439,7 +439,7 @@
     }
    ],
    "source": [
-    "x.index_corr().plot();   # expanding correlation rolling_window = None"
+    "x.index_corr().plot();  # expanding correlation rolling_window = None"
    ]
   },
   {

main.py

@@ -5,7 +5,14 @@
 # ef = ok.EfficientFrontier(indexes, ccy="RUB", full_frontier=True, inflation=False, n_points=50)
 # ef.plot_cml(rf_return=0.15, y_axe="cagr")
 # plt.show()
+#
+# pf = ok.Portfolio(["SPY.US", "BND.US"], weights=[.5, .5], rebalancing_period="month")
+# pf.wealth_index_with_assets.plot()
+# plt.show()
+
+i = ok.Inflation("RUB.INFL")
+
+i.set_values_monthly(date="2023-12", value=0.0122)
 
-pf = ok.Portfolio(["SPY.US", "BND.US"], weights=[.5, .5], rebalancing_period="monthly")
-pf.wealth_index_with_assets.plot()
-plt.show()
+print(i.values_monthly)
+print(i)

okama/common/helpers/helpers.py

@@ -359,26 +359,18 @@ class Rebalance:
     """
     Methods for rebalancing portfolio.
     """
+
     # From Pandas resamples alias: https://pandas.pydata.org/docs/user_guide/timeseries.html#timeseries-offset-aliases
-    frequency_mapping = {
-        "none": "none",
-        "annually": "Y",
-        "semi-annually": "2Q",
-        "quarterly": "Q",
-        "monthly": "M"
-    }
-
-    def __init__(self,
-                 period: str = "annually",
-                 abs_deviation: Optional[float] = None,
-                 rel_deviation: Optional[float] = None):
+    frequency_mapping = {"none": "none", "year": "Y", "half-year": "2Q", "quarter": "Q", "month": "M"}
+
+    def __init__(
+        self, period: str = "year", abs_deviation: Optional[float] = None, rel_deviation: Optional[float] = None
+    ):
         self.period = period
         self.abs_deviation = abs_deviation
         self.rel_deviation = rel_deviation
         self.pandas_frequency = self.frequency_mapping.get(self.period)
 
-
-
     def wealth_ts(self, weights: list, ror: pd.DataFrame) -> pd.Series:
         """
         Calculate wealth index time series of rebalanced portfolio given returns time series of the assets.

okama/frontier/multi_period.py

@@ -228,7 +228,7 @@ def gmv_monthly_weights(self) -> np.ndarray:
 
         # Set the objective function
         def objective_function(w):
-            risk = helpers.Rebalance.return_ts(w, ror, period=period).std()
+            risk = helpers.Rebalance(period=period).return_ts(w, ror).std()
             return risk
 
         # construct the constraints
@@ -272,7 +272,7 @@ def gmv_annual_weights(self) -> np.ndarray:
 
         # Set the objective function
         def objective_function(w):
-            ts = helpers.Rebalance.return_ts(w, ror, period=period)
+            ts = helpers.Rebalance(period=period).return_ts(w, ror)
             mean_return = ts.mean()
             risk = ts.std()
             return helpers.Float.annualize_risk(risk=risk, mean_return=mean_return)
@@ -296,16 +296,18 @@ def _get_gmv_monthly(self) -> Tuple[float, float]:
         Global Minimum Volatility portfolio is a portfolio with the lowest risk of all possible.
         """
         return (
-            helpers.Rebalance.return_ts(
+            helpers.Rebalance(period=self.rebalancing_period)
+            .return_ts(
                 self.gmv_monthly_weights,
                 self.assets_ror,
-                period=self.rebalancing_period,
-            ).std(),
-            helpers.Rebalance.return_ts(
+            )
+            .std(),
+            helpers.Rebalance(period=self.rebalancing_period)
+            .return_ts(
                 self.gmv_monthly_weights,
                 self.assets_ror,
-                period=self.rebalancing_period,
-            ).mean(),
+            )
+            .mean(),
         )
 
     @property
@@ -330,7 +332,7 @@ def gmv_annual_values(self) -> Tuple[float, float]:
         >>> frontier.gmv_annual_values
         (0.03695845106087943, 0.04418318557516887)
         """
-        returns = helpers.Rebalance.return_ts(self.gmv_annual_weights, self.assets_ror, period=self.rebalancing_period)
+        returns = helpers.Rebalance(period=self.rebalancing_period).return_ts(self.gmv_annual_weights, self.assets_ror)
         return (
             helpers.Float.annualize_risk(returns.std(), returns.mean()),
             (returns + 1.0).prod() ** (settings._MONTHS_PER_YEAR / returns.shape[0]) - 1.0,
@@ -367,7 +369,7 @@ def global_max_return_portfolio(self) -> dict:
         # Set the objective function
         def objective_function(w):
             # Accumulated return for rebalanced portfolio time series
-            objective_function.returns = helpers.Rebalance.return_ts(w, ror, period=period)
+            objective_function.returns = helpers.Rebalance(period=period).return_ts(w, ror)
             accumulated_return = (objective_function.returns + 1.0).prod() - 1.0
             return -accumulated_return
 
@@ -393,7 +395,7 @@ def objective_function(w):
         return point
 
     def _get_cagr(self, weights):
-        ts = helpers.Rebalance.return_ts(weights, self.assets_ror, period=self.rebalancing_period)
+        ts = helpers.Rebalance(period=self.rebalancing_period).return_ts(weights, self.assets_ror)
         acc_return = (ts + 1.0).prod() - 1.0
         return (1.0 + acc_return) ** (settings._MONTHS_PER_YEAR / ts.shape[0]) - 1.0
 
@@ -424,7 +426,7 @@ def minimize_risk(self, target_value: float) -> Dict[str, float]:
 
         def objective_function(w):
             # annual risk
-            ts = helpers.Rebalance.return_ts(w, self.assets_ror, period=self.rebalancing_period)
+            ts = helpers.Rebalance(period=self.rebalancing_period).return_ts(w, self.assets_ror)
             risk_monthly = ts.std()
             mean_return = ts.mean()
             return helpers.Float.annualize_risk(risk_monthly, mean_return)
@@ -480,7 +482,7 @@ def _maximize_risk(self, target_return: float) -> Dict[str, float]:
 
         def objective_function(w):
             # annual risk
-            ts = helpers.Rebalance.return_ts(w, self.assets_ror, period=self.rebalancing_period)
+            ts = helpers.Rebalance(period=self.rebalancing_period).return_ts(w, self.assets_ror)
             risk_monthly = ts.std()
             mean_return = ts.mean()
             result = -helpers.Float.annualize_risk(risk_monthly, mean_return)
@@ -797,9 +799,8 @@ def get_monte_carlo(self, n: int = 100) -> pd.DataFrame:
 
         # Portfolio risk and cagr for each set of weights
         portfolios_ror = weights_df.aggregate(
-            helpers.Rebalance.return_ts,
+            helpers.Rebalance(period=self.rebalancing_period).return_ts,
             ror=self.assets_ror,
-            period=self.rebalancing_period,
         )
         random_portfolios = pd.DataFrame()
         for _, data in portfolios_ror.iterrows():

okama/frontier/single_period.py

@@ -891,7 +891,7 @@ def get_monte_carlo(self, n: int = 100, kind: str = "mean") -> pd.DataFrame:
             random_portfolios = helpers.Frame.change_columns_order(random_portfolios, ["Risk", second_column])
         return random_portfolios
 
-    def plot_transition_map(self, x_axe: str = 'risk', figsize: Optional[tuple] = None) -> plt.axes:
+    def plot_transition_map(self, x_axe: str = "risk", figsize: Optional[tuple] = None) -> plt.axes:
         """
         Plot Transition Map for optimized portfolios on the single period Efficient Frontier.
 
@@ -941,10 +941,10 @@ def plot_transition_map(self, x_axe: str = 'risk', figsize: Optional[tuple] = No
         """
         ef = self.ef_points
         linestyle = itertools.cycle(("-", "--", ":", "-."))
-        if x_axe.lower() == 'cagr':
+        if x_axe.lower() == "cagr":
             xlabel = "CAGR (Compound Annual Growth Rate)"
             x_axe = "CAGR"
-        elif x_axe.lower() == 'risk':
+        elif x_axe.lower() == "risk":
             xlabel = "Risk (volatility)"
             x_axe = "Risk"
         else:

okama/macro.py

@@ -4,6 +4,7 @@
 import numpy as np
 import pandas as pd
 
+import okama.common.validators
 from okama import settings
 from okama.api import data_queries, namespaces
 from okama.common.helpers import helpers
@@ -36,12 +37,9 @@ def __init__(
         self._get_symbol_data(symbol)
         self._first_date = first_date
         self._last_date = last_date
-        self.first_date: pd.Timestamp = self.values_monthly.index[0].to_timestamp()
-        self.last_date: pd.Timestamp = self.values_monthly.index[-1].to_timestamp()
-        self.pl = settings.PeriodLength(
-            self.values_monthly.shape[0] // settings._MONTHS_PER_YEAR,
-            self.values_monthly.shape[0] % settings._MONTHS_PER_YEAR,
-        )
+        self._values_monthly = self._get_values_monthly()
+        self._set_first_last_dates()
+
         self._pl_txt = f"{self.pl.years} years, {self.pl.months} months"
 
     def __repr__(self):
@@ -60,6 +58,17 @@ def __repr__(self):
     def _check_namespace(self):
         pass
 
+    def _set_first_last_dates(self):
+        self.first_date: pd.Timestamp = self.values_monthly.index[0].to_timestamp()
+        self.last_date: pd.Timestamp = self.values_monthly.index[-1].to_timestamp()
+        self.pl = settings.PeriodLength(
+            self.values_monthly.shape[0] // settings._MONTHS_PER_YEAR,
+            self.values_monthly.shape[0] % settings._MONTHS_PER_YEAR,
+        )
+
+    def _get_values_monthly(self) -> pd.Series:
+        return data_queries.QueryData.get_macro_ts(self.symbol, self._first_date, self._last_date, period="M")
+
     def _get_symbol_data(self, symbol) -> None:
         x = data_queries.QueryData.get_symbol_info(symbol)
         self.ticker: str = x["code"]
@@ -78,7 +87,17 @@ def values_monthly(self) -> pd.Series:
         Series
             Time series of values historical data (monthly).
         """
-        return data_queries.QueryData.get_macro_ts(self.symbol, self._first_date, self._last_date, period="M")
+        return self._values_monthly
+
+    def set_values_monthly(self, date: str, value: float):
+        """
+        Set monthly value for the past or future date.
+
+        The date should be in month period format ("2023-12").
+        """
+        okama.common.validators.validate_real("value", value)
+        self._values_monthly[pd.Period(date, freq='M')] = value
+        self._set_first_last_dates()
 
     def describe(self, years: Tuple[int, ...] = (1, 5, 10)) -> pd.DataFrame:
         """

okama/portfolio.py

@@ -12,7 +12,6 @@
 from okama.common.helpers.helpers import Rebalance
 
 
-
 class Portfolio(make_asset_list.ListMaker):
     """
     Implementation of investment portfolio.
@@ -56,7 +55,7 @@ class Portfolio(make_asset_list.ListMaker):
         The weight of an asset is the percent of an investment portfolio that corresponds to the asset.
         If weights = None an equally weighted portfolio is created (all weights are equal).
 
-    rebalancing_period : {'month', 'year', 'none'}, default 'month'
+    rebalancing_period : {'none', 'month', 'quarter', 'half-year', 'year'}, default 'month'
         Rebalancing period (rebalancing frequency) is predetermined time intervals when
         the investor rebalances the portfolio. If 'none' assets weights are not rebalanced.
 
@@ -176,9 +175,8 @@ def weights_ts(self) -> pd.DataFrame:
         >>> plt.show()
         """
         if self.rebalancing_period != "month":
-            return helpers.Rebalance.assets_weights_ts(
+            return helpers.Rebalance(period=self.rebalancing_period).assets_weights_ts(
                 ror=self.assets_ror,
-                period=self.rebalancing_period,
                 weights=self.weights,
             )
         values = np.tile(self.weights, (self.ror.shape[0], 1))
@@ -207,7 +205,7 @@ def rebalancing_period(self, rebalancing_period: str):
         if rebalancing_period in Rebalance.frequency_mapping.keys():
             self._rebalancing_period = rebalancing_period
         else:
-            raise ValueError(f'rebalancing_period must be in {Rebalance.frequency_mapping.keys()}')
+            raise ValueError(f"rebalancing_period must be in {Rebalance.frequency_mapping.keys()}")
 
     @property
     def symbol(self) -> str:

pyproject.toml

@@ -35,15 +35,9 @@ matplotlib = "^3.5.1"
 requests = "<=2.31.0"   # Windows has an issue with 2.32.0
 joblib = "^1.1.0"
 
-[tool.poetry.group.test]
-optional = true
-
 [tool.poetry.group.test.dependencies]
 pytest = "^6.0.0"
-black = "^22.3.0"
-
-[tool.poetry.group.docs]
-optional = true
+black = "^23.11.0"
 
 [tool.poetry.group.docs.dependencies]
 sphinx = "^5.0.0"
@@ -55,9 +49,6 @@ pandoc = "^2.2"
 recommonmark = "^0.7.1"
 Jinja2 = "3.0.3"
 
-[tool.poetry.group.jupyter]
-optional = true
-
 [tool.poetry.group.jupyter.dependencies]
 jupyter = "^1.0.0"
 ipykernel = "^6.15.0"

tests/test_asset_list.py

@@ -67,7 +67,7 @@ def test_currencies(self):
             "asset list": "USD",
         }
         assert self.currencies.names == {
-            "RUBUSD.FX": "Russian Rouble/US Dollar FX Cross Rate",
+            "RUBUSD.FX": "Central Bank of Russia rate for RUBUSD (US Dollar)",
             "EURUSD.FX": "EURUSD",
             "CNYUSD.FX": "Chinese Renminbi/US Dollar FX Cross Rate",
         }