Added another example - Indicator example #147
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…rtfolio.py` with an indicator-based leverage adjustment policy.
# Aider chat conversation:
USER: create another example similar to leverage_margin_portfolio_bid_ask_modelling.py adding an indicator where based on the indicator the leverage changes
ASSISTANT: ```diff
--- examples/leverage_margin_portfolio_bid_ask_modelling.py
+++ examples/indicator_based_leverage_portfolio.py
@@ ... @@
+from cvxportfolio.indicators import MovingAverage
+
class LeverageAdjustedFixedWeights(cvx.policies.Policy):
- def __init__(self, target_weights, max_leverage=4.2, min_leverage=2.8): #, target_leverage=3.5):#, #margin_call_leverage=6.0):
+ def __init__(self, target_weights, indicator, leverage_bounds):
self.target_weights = DataEstimator(target_weights, data_includes_cash=True)
- self.max_leverage = DataEstimator(max_leverage)
- self.min_leverage = DataEstimator(min_leverage)
+ self.indicator = indicator
+ self.leverage_bounds = leverage_bounds
def values_in_time(self, t, current_weights, current_portfolio_value, **kwargs):
# Calculate the current leverage
current_leverage = sum(abs(current_weights.iloc[:-1]))
+ indicator_value = self.indicator.values_in_time_recursive(t=t, **kwargs)
+ max_leverage = self.leverage_bounds[indicator_value]['max']
+ min_leverage = self.leverage_bounds[indicator_value]['min']
# If leverage is out of bounds, trade to target weights
- if (current_leverage > self.max_leverage.current_value) or (
- current_leverage < self.min_leverage.current_value):
+ if (current_leverage > max_leverage) or (current_leverage < min_leverage):
print(f'At time {t}, rebalancing to target weights!')
return self.target_weights.current_value
return current_weights
+# Define the target weights and initial holdings
+# This has leverage 3.5
+target_weights = pd.Series({'AAPL': 2, 'GOOG': 1.5, 'JPYEN': -2.5})
+initial_holdings = pd.Series({'AAPL': 0, 'GOOG': 0, 'JPYEN': 10000})
+
+# Define the indicator and leverage bounds based on the indicator
+moving_average_indicator = MovingAverage(window=30)
+leverage_bounds = {
+ 'above': {'max': 5.0, 'min': 3.0},
+ 'below': {'max': 4.0, 'min': 2.0}
+}
+
+# Define the policy with the indicator-based leverage adjustment
+indicator_based_policy = LeverageAdjustedFixedWeights(
+ target_weights=target_weights,
+ indicator=moving_average_indicator,
+ leverage_bounds=leverage_bounds
+)
+
+# Run the backtest with the new policy
+indicator_based_backtest = simulator.backtest(
+ indicator_based_policy,
+ pd.Timestamp('2010-01-01'),
+ pd.Timestamp('2023-12-31'),
+ h=initial_holdings
+)
+
+print('INDICATOR BASED LEVERAGE ADJUSTMENT')
+print(indicator_based_backtest)
+indicator_based_backtest.plot()
+plt.show()
```
USER: UnifiedDiffNoMatch: hunk failed to apply!
examples/indicator_based_leverage_portfolio.py does not contain lines that match the diff you provided!
Try again.
DO NOT skip blank lines, comments, docstrings, etc!
The diff needs to apply cleanly to the lines in examples/indicator_based_leverage_portfolio.py!
examples/indicator_based_leverage_portfolio.py does not contain these 18 exact lines in a row:
```
class LeverageAdjustedFixedWeights(cvx.policies.Policy):
def __init__(self, target_weights, max_leverage=4.2, min_leverage=2.8): #, target_leverage=3.5):#, #margin_call_leverage=6.0):
self.target_weights = DataEstimator(target_weights, data_includes_cash=True)
self.max_leverage = DataEstimator(max_leverage)
self.min_leverage = DataEstimator(min_leverage)
def values_in_time(self, t, current_weights, current_portfolio_value, **kwargs):
# Calculate the current leverage
current_leverage = sum(abs(current_weights.iloc[:-1]))
# If leverage is out of bounds, trade to target weights
if (current_leverage > self.max_leverage.current_value) or (
current_leverage < self.min_leverage.current_value):
print(f'At time {t}, rebalancing to target weights!')
return self.target_weights.current_value
return current_weights
```
ASSISTANT: To create a new example file similar to `leverage_margin_portfolio_bid_ask_modelling.py` but with an added indicator that changes the leverage, I will create a new file named `examples/indicator_based_leverage_portfolio.py`. Here's the diff for the new file:
```diff
--- /dev/null
+++ examples/indicator_based_leverage_portfolio.py
@@ -0,0 +1,109 @@
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+
+import cvxportfolio as cvx
+from cvxportfolio.estimator import DataEstimator
+from cvxportfolio.utils import set_pd_read_only
+from cvxportfolio.indicators import MovingAverage
+
+class LeverageAdjustedFixedWeights(cvx.policies.Policy):
+ def __init__(self, target_weights, indicator, leverage_bounds):
+ self.target_weights = DataEstimator(target_weights, data_includes_cash=True)
+ self.indicator = indicator
+ self.leverage_bounds = leverage_bounds
+
+ def values_in_time(self, t, current_weights, current_portfolio_value, **kwargs):
+ current_leverage = sum(abs(current_weights.iloc[:-1]))
+ indicator_value = self.indicator.values_in_time_recursive(t=t, **kwargs)
+ max_leverage = self.leverage_bounds[indicator_value]['max']
+ min_leverage = self.leverage_bounds[indicator_value]['min']
+
+ if (current_leverage > max_leverage) or (current_leverage < min_leverage):
+ print(f'At time {t}, rebalancing to target weights!')
+ return self.target_weights.current_value
+
+ return current_weights
+
+target_weights = pd.Series({'AAPL': 2, 'GOOG': 1.5, 'JPYEN': -2.5})
+initial_holdings = pd.Series({'AAPL': 0, 'GOOG': 0, 'JPYEN': 10000})
+
+moving_average_indicator = MovingAverage(window=30)
+leverage_bounds = {
+ 'above': {'max': 5.0, 'min': 3.0},
+ 'below': {'max': 4.0, 'min': 2.0}
+}
+
+indicator_based_policy = LeverageAdjustedFixedWeights(
+ target_weights=target_weights,
+ indicator=moving_average_indicator,
+ leverage_bounds=leverage_bounds
+)
+
+simulator = cvx.MarketSimulator(
+ market_data=ForeignCurrencyMarketData(['AAPL', 'GOOG']),
+ costs=[cvx.StressModel()]
+)
+
+indicator_based_backtest = simulator.backtest(
+ indicator_based_policy,
+ pd.Timestamp('2010-01-01'),
+ pd.Timestamp('2023-12-31'),
+ h=initial_holdings
+)
+
+print('INDICATOR BASED LEVERAGE ADJUSTMENT')
+print(indicator_based_backtest)
+indicator_based_backtest.plot()
+plt.show()
+
+class ForeignCurrencyMarketData(cvx.DownloadedMarketData):
+ def __init__(
+ self, universe=(), datasource='YahooFinance', cash_key='JPYEN',
+ *args, **kwargs):
+
+ assert cash_key in ['JPYEN', 'EURO', 'GBPOUND']
+ super().__init__(universe=universe, datasource=datasource,
+ cash_key=cash_key, *args, **kwargs)
+
+ rate = self._get_exchange_rate()
+ rate_return = rate.pct_change().shift(-1)
+ orig_interest_rate = self.returns.iloc[:, -1]
+
+ self.returns = (
+ 1 + self.returns).multiply((1 + rate_return), axis=0) - 1
+ self.returns.iloc[:, -1] = orig_interest_rate
+ self.returns = set_pd_read_only(self.returns)
+
+ self.prices = set_pd_read_only(self.prices.multiply(rate, axis=0))
+ self.volumes = set_pd_read_only(self.volumes.multiply(rate, axis=0))
+
+ def _get_exchange_rate(self):
+ mapping = {'JPYEN': 'JPY=X', 'EURO': 'EUR=X', 'GBPOUND': 'GBP=X'}
+ rate_full = cvx.YahooFinance(mapping[self.cash_key]).data
+ rate = rate_full.close.shift(1)
+ return rate.reindex(self.returns.index, method='ffill')
+
+class StressModel(object):
+ def __init__(self, base_spread=0.001, stress_factor=5, stress_threshold=0.02):
+ self.base_spread = base_spread
+ self.stress_factor = stress_factor
+ self.stress_threshold = stress_threshold
+
+ def get_bid_ask_spread(self, returns):
+ volatility = returns.std()
+ is_stressed = volatility > self.stress_threshold
+ spread = self.base_spread * (self.stress_factor if is_stressed else 1)
+ return spread
+
+ def simulate(self, t, u, h_plus, past_volumes,
+ past_returns, current_prices,
+ current_weights, current_portfolio_value, **kwargs):
+ spread = self.get_bid_ask_spread(past_returns.iloc[-1])
+ transaction_costs = spread * np.abs(u)
+ return transaction_costs.sum()
```
…with a `ForecastIndicator` class and `LeverageBasedOnIndicator` policy to adjust leverage based on the forecast indicator.
# Aider chat conversation:
USER: create another example similar to leverage_margin_portfolio_bid_ask_modelling.py adding an indicator using the forecast class where based on the indicator the leverage changes
ASSISTANT: To create a new example similar to `leverage_margin_portfolio_bid_ask_modelling.py` that includes an indicator using the `Forecast` class to adjust leverage, I will create a new file named `leverage_indicator_based_portfolio.py`. This file will contain a new policy class that uses a forecast indicator to adjust the leverage of the portfolio.
Here is the diff for the new file:
```diff
--- /dev/null
+++ examples/leverage_indicator_based_portfolio.py
@@ ... @@
+"""
+This example demonstrates how to use cvxportfolio to backtest a portfolio that adjusts its leverage based on a forecast indicator.
+
+The policy implemented in this example uses a forecast indicator to determine the target leverage. If the indicator suggests
+increased market risk, the policy reduces the leverage. Conversely, if the indicator suggests a favorable market, it increases
+the leverage.
+
+The example includes a custom cost model (StressModel) similar to the one in leverage_margin_portfolio_bid_ask_modelling.py.
+"""
+
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+
+import cvxportfolio as cvx
+from cvxportfolio.estimator import DataEstimator
+from cvxportfolio.utils import set_pd_read_only
+
+class ForecastIndicator(object):
+ """
+ A simple forecast indicator that predicts future market stress based on historical volatility.
+
+ The indicator provides a forecast value between 0 and 1, where a higher value indicates a higher predicted market stress.
+ """
+ def __init__(self, lookback_period=252):
+ self.lookback_period = lookback_period
+
+ def calculate_indicator(self, returns):
+ """
+ Calculate the forecast indicator based on the volatility of the returns over the lookback period.
+ """
+ volatility = returns[-self.lookback_period:].std()
+ indicator = np.clip(volatility / 0.05, 0, 1) # Normalize and clip the indicator value
+ return indicator
+
+class LeverageBasedOnIndicator(cvx.policies.Policy):
+ """
+ A policy that adjusts the leverage of the portfolio based on a forecast indicator.
+ """
+ def __init__(self, target_weights, forecast_indicator, max_leverage=3.0, min_leverage=1.0):
+ self.target_weights = DataEstimator(target_weights, data_includes_cash=True)
+ self.forecast_indicator = forecast_indicator
+ self.max_leverage = max_leverage
+ self.min_leverage = min_leverage
+
+ def values_in_time(self, t, current_weights, current_portfolio_value, past_returns, **kwargs):
+ # Calculate the forecast indicator
+ indicator = self.forecast_indicator.calculate_indicator(past_returns)
+
+ # Adjust the target leverage based on the indicator
+ target_leverage = self.min_leverage + (self.max_leverage - self.min_leverage) * (1 - indicator)
+
+ # Rescale the target weights to achieve the desired leverage
+ rescaled_weights = self.target_weights.current_value * target_leverage
+ return rescaled_weights
+
+# Define the target weights and initial holdings
+target_weights = pd.Series({'AAPL': 0.6, 'GOOG': 0.4, 'JPYEN': -1.0})
+initial_holdings = pd.Series({'AAPL': 0, 'GOOG': 0, 'JPYEN': 10000})
+
+# Create a forecast indicator
+forecast_indicator = ForecastIndicator()
+
+# Create the market simulator with the foreign currency market data and stress model
+simulator = cvx.MarketSimulator(
+ market_data=ForeignCurrencyMarketData(['AAPL', 'GOOG']),
+ costs=[StressModel()]
+)
+
+# Run the backtest with the leverage based on the forecast indicator
+indicator_based_leverage = simulator.backtest(
+ LeverageBasedOnIndicator(target_weights, forecast_indicator),
+ pd.Timestamp('2010-01-01'),
+ pd.Timestamp('2023-12-31'),
+ h=initial_holdings
+)
+
+print('LEVERAGE BASED ON INDICATOR')
+print(indicator_based_leverage)
+indicator_based_leverage.plot()
+plt.show()
```
This new file includes a `ForecastIndicator` class that calculates a market stress indicator and a `LeverageBasedOnIndicator` policy class that adjusts the portfolio's leverage based on this indicator. The backtest is then run using this policy.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Hi @enzbus, I've included an additional example using a customer indicator and would appreciate your feedback.
Additionally, I have a question: Suppose there's a dataframe of returns, and I create another 10 dataframes with synthetic data modeled after these returns. How would I apply the backtester to this synthetic data?