Update method to download historical data in Yahoo Finance

calculate_expected_volatility.py

@@ -0,0 +1,41 @@
+import numpy as np
+import math
+
+# Mock data from the test
+prices = [
+    100.0,  # 2020-02-15
+    98.5,   # 2020-02-14
+    99.2,   # 2020-02-13
+    97.8,   # 2020-02-12
+    98.1,   # 2020-02-11
+    99.5,   # 2020-02-10
+    100.2,  # 2020-02-09
+    99.8,   # 2020-02-08
+    100.5,  # 2020-02-07
+    101.2,  # 2020-02-06
+    100.8,  # 2020-02-05
+    101.5,  # 2020-02-04
+    102.0,  # 2020-02-03
+    101.8,  # 2020-02-02
+    102.2,  # 2020-02-01
+    101.9,  # 2020-01-31
+    102.5,  # 2020-01-30
+    103.0,  # 2020-01-29
+    102.8,  # 2020-01-28
+    103.2,  # 2020-01-27
+    103.5   # 2020-01-26
+]
+
+# Calculate log returns
+log_returns = []
+for i in range(20):  # window_size = 20
+    log_returns.append(math.log(prices[i] / prices[i+1]))
+
+# Calculate volatility
+volatility = np.std(log_returns, ddof=1) * np.sqrt(252)  # annualized volatility
+
+# Calculate performance (prices[0] / prices[window_size] - 1.0)
+performance = prices[0] / prices[20] - 1.0
+
+print(f"Calculated volatility: {volatility:.6f}")
+print(f"Performance: {performance:.6f}") 

inverse_volatility.py

@@ -1,13 +1,13 @@
 #!/usr/local/bin/python3
 
-# Author: Zebing Lin (https://github.com/linzebing)
+# Adapted from: Zebing Lin (https://github.com/linzebing)
 
 from datetime import datetime, date
 import math
 import numpy as np
 import time
 import sys
-import requests
+import yfinance as yf
 
 if len(sys.argv) == 1:
     symbols = ['UPRO', 'TMF']
@@ -19,39 +19,50 @@
 num_trading_days_per_year = 252
 window_size = 20
 date_format = "%Y-%m-%d"
-end_timestamp = int(time.time())
+reference_date = datetime.today()
+end_timestamp = int(reference_date.timestamp())
 start_timestamp = int(end_timestamp - (1.4 * (window_size + 1) + 4) * 86400)
 
-
-def get_volatility_and_performance(symbol):
-    download_url = "https://query1.finance.yahoo.com/v7/finance/download/{}?period1={}&period2={}&interval=1d&events=history&crumb=a7pcO//zvcW".format(symbol, start_timestamp, end_timestamp)
-    lines = requests.get(download_url, cookies={'B': 'chjes25epq9b6&b=3&s=18'}).text.strip().split('\n')
-    assert lines[0].split(',')[0] == 'Date'
-    assert lines[0].split(',')[4] == 'Close'
-    prices = []
-    for line in lines[1:]:
-        prices.append(float(line.split(',')[4]))
+def get_volatility_and_performance(symbol, reference_date=reference_date):
+    # Get data using yfinance
+    ticker = yf.Ticker(symbol)
+    hist = ticker.history(start=datetime.fromtimestamp(start_timestamp), end=datetime.fromtimestamp(end_timestamp))
+
+    if len(hist) < window_size + 1:
+        raise Exception(f"Not enough data points for {symbol}")
+
+    # Get closing prices
+    prices = hist['Close'].values.tolist()
     prices.reverse()
+
     volatilities_in_window = []
-
     for i in range(window_size):
         volatilities_in_window.append(math.log(prices[i] / prices[i+1]))
-
-    most_recent_date = datetime.strptime(lines[-1].split(',')[0], date_format).date()
-    assert (date.today() - most_recent_date).days <= 4, "today is {}, most recent trading day is {}".format(date.today(), most_recent_date)
-
-    return np.std(volatilities_in_window, ddof = 1) * np.sqrt(num_trading_days_per_year), prices[0] / prices[window_size] - 1.0
-
-volatilities = []
-performances = []
-sum_inverse_volatility = 0.0
-for symbol in symbols:
-    volatility, performance = get_volatility_and_performance(symbol)
-    sum_inverse_volatility += 1 / volatility
-    volatilities.append(volatility)
-    performances.append(performance)
-
-print ("Portfolio: {}, as of {} (window size is {} days)".format(str(symbols), date.today().strftime('%Y-%m-%d'), window_size))
-for i in range(len(symbols)):
-    print ('{} allocation ratio: {:.2f}% (anualized volatility: {:.2f}%, performance: {:.2f}%)'.format(symbols[i], float(100 / (volatilities[i] * sum_inverse_volatility)), float(volatilities[i] * 100), float(performances[i] * 100)))
+
+    most_recent_date = hist.index[-1].date()
+    assert (reference_date.date() - most_recent_date).days <= 4, f"reference date is {reference_date.date()}, most recent trading day is {most_recent_date}"
+
+    return np.std(volatilities_in_window, ddof=1) * np.sqrt(num_trading_days_per_year), prices[0] / prices[window_size] - 1.0
+
+def main():
+    volatilities = []
+    performances = []
+    sum_inverse_volatility = 0.0
+    for symbol in symbols:
+        volatility, performance = get_volatility_and_performance(symbol, reference_date)
+        sum_inverse_volatility += 1 / volatility
+        volatilities.append(volatility)
+        performances.append(performance)
+
+    print("Portfolio: {}, as of {} (window size is {} days)".format(str(symbols), reference_date.strftime('%Y-%m-%d'), window_size))
+    for i in range(len(symbols)):
+        print('{} allocation ratio: {:.2f}% (anualized volatility: {:.2f}%, performance: {:.2f}%)'.format(
+            symbols[i], 
+            float(100 / (volatilities[i] * sum_inverse_volatility)), 
+            float(volatilities[i] * 100), 
+            float(performances[i] * 100)
+        ))
+
+if __name__ == "__main__":
+    main()
 

risk_parity.py

@@ -0,0 +1,59 @@
+#!/usr/local/bin/python3
+
+import numpy as np
+import pandas as pd
+import yfinance as yf
+from scipy.optimize import minimize
+import matplotlib.pyplot as plt
+
+tickers = ['VTV', 'BRK-B', 'ARKK']  # S&P 500, 20+ Year Treasury Bond, Gold
+data = yf.download(tickers, start="2015-05-22", end="2025-06-12")['Close']
+returns = data.pct_change().dropna()
+
+def calculate_portfolio_variance(weights, cov_matrix):
+    return np.dot(weights.T, np.dot(cov_matrix, weights))
+
+def calculate_risk_contribution(weights, cov_matrix):
+    portfolio_variance = calculate_portfolio_variance(weights, cov_matrix)
+    marginal_contrib = np.dot(cov_matrix, weights)
+    risk_contrib = np.multiply(weights, marginal_contrib) / portfolio_variance
+    return risk_contrib
+
+def risk_parity_objective(weights, cov_matrix):
+    risk_contrib = calculate_risk_contribution(weights, cov_matrix)
+    target_risk = np.mean(risk_contrib)
+    return np.sum((risk_contrib - target_risk) ** 2)
+
+def get_risk_parity_weights(cov_matrix):
+    num_assets = len(cov_matrix)
+    initial_weights = np.ones(num_assets) / num_assets
+    constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
+    bounds = [(0, 1) for _ in range(num_assets)]
+
+    result = minimize(risk_parity_objective, initial_weights, args=(cov_matrix,), 
+                      method='SLSQP', bounds=bounds, constraints=constraints)
+
+    return result.x
+
+cov_matrix = returns.cov()
+risk_parity_weights = get_risk_parity_weights(cov_matrix)
+print("Risk Parity Weights:", risk_parity_weights)
+
+plt.figure(figsize=(8, 6))
+plt.bar(tickers, risk_parity_weights, color='skyblue')
+plt.title('Risk Parity Portfolio Allocation')
+plt.xlabel('Assets')
+plt.ylabel('Weights')
+plt.show()
+
+risk_parity_returns = np.dot(returns, risk_parity_weights)
+equal_weighted_returns = np.dot(returns, np.ones(len(tickers)) / len(tickers))
+
+plt.figure(figsize=(10, 6))
+plt.plot((1 + risk_parity_returns).cumprod(), label='Risk Parity Portfolio')
+plt.plot((1 + equal_weighted_returns).cumprod(), label='Equal Weighted Portfolio', linestyle='--')
+plt.title('Portfolio Performance Comparison')
+plt.xlabel('Date')
+plt.ylabel('Cumulative Returns')
+plt.legend()
+plt.show()

test_inverse_volatility.py

@@ -0,0 +1,73 @@
+import unittest
+from datetime import datetime, timedelta
+from unittest.mock import patch, MagicMock
+import numpy as np
+import pandas as pd
+from inverse_volatility import get_volatility_and_performance
+
+class TestInverseVolatility(unittest.TestCase):
+    def setUp(self):
+        # Mock data for UPRO around 2020-02-15
+        self.mock_prices = [
+            103.5,   # 2020-01-26 (oldest)
+            103.2,  # 2020-01-27
+            102.8,  # 2020-01-28
+            103.0,  # 2020-01-29
+            102.5,  # 2020-01-30
+            101.9,  # 2020-01-31
+            102.2,  # 2020-02-01
+            101.8,  # 2020-02-02
+            102.0,  # 2020-02-03
+            101.5,  # 2020-02-04
+            100.8,  # 2020-02-05
+            101.2,  # 2020-02-06
+            100.5,  # 2020-02-07
+            99.8,   # 2020-02-08
+            100.2,  # 2020-02-09
+            99.5,   # 2020-02-10
+            98.1,   # 2020-02-11
+            97.8,   # 2020-02-12
+            99.2,   # 2020-02-13
+            98.5,   # 2020-02-14
+            100.0   # 2020-02-15 (most recent)
+        ]
+        # The most recent date is 2020-02-15, matching the reference date
+        self.mock_dates = [datetime(2020, 1, 26) + timedelta(days=i) for i in range(21)]
+        self.mock_df = pd.DataFrame({
+            'Close': self.mock_prices
+        }, index=self.mock_dates)
+
+    @patch('yfinance.Ticker')
+    def test_get_volatility_and_performance(self, mock_ticker):
+        # Setup mock
+        mock_instance = MagicMock()
+        mock_instance.history.return_value = self.mock_df
+        mock_ticker.return_value = mock_instance
+
+        # Test with UPRO
+        volatility, performance = get_volatility_and_performance('UPRO')
+
+        # Expected values (calculated from mock data)
+        expected_volatility = 0.114866  # Calculated volatility
+        expected_performance = -0.033816  # Calculated performance
+
+        # Assert with some tolerance for floating point calculations
+        self.assertAlmostEqual(volatility, expected_volatility, places=6)
+        self.assertAlmostEqual(performance, expected_performance, places=6)
+
+    @patch('yfinance.Ticker')
+    def test_insufficient_data(self, mock_ticker):
+        # Create mock history with insufficient data
+        insufficient_df = self.mock_df.iloc[:10]  # Only 10 days of data
+        mock_instance = MagicMock()
+        mock_instance.history.return_value = insufficient_df
+        mock_ticker.return_value = mock_instance
+
+        # Test that it raises an exception for insufficient data
+        with self.assertRaises(Exception) as context:
+            get_volatility_and_performance('UPRO')
+
+        self.assertTrue('Not enough data points' in str(context.exception))
+
+if __name__ == '__main__':
+    unittest.main()