A merry & shrewd investing community |
Best Of MI | Best Of | Favourites & Replies | All Boards | Post of the Week! |
![]() Best Of MI | Best Of | Favourites & Replies | All Boards | Post of the Week! |
Unthreaded | Threaded | Whole Thread (8) | Prev | Next |
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime
def get_historical_data(ticker, start_date, end_date):
"""Fetch historical data for a given ticker"""
stock = yf.Ticker(ticker)
df = stock.history(start=start_date, end=end_date)
return df['Close']
def calculate_signals(spy_prices):
"""Calculate trading signals based on 52-week lows and 26-week highs"""
# Calculate rolling windows
rolling_52_week_low = spy_prices.rolling(window=252).min() # 252 trading days ≈ 52 weeks
rolling_26_week_high = spy_prices.rolling(window=126).max() # 126 trading days ≈ 26 weeks
# Initialize signals DataFrame
signals = pd.DataFrame(index=spy_prices.index)
signals['spy_price'] = spy_prices
signals['52_week_low'] = rolling_52_week_low
signals['26_week_high'] = rolling_26_week_high
# Generate buy/sell signals
signals['position'] = 'SPY' # Start with SPY position
# Initialize with first position as SPY
current_position = 'SPY'
positions = []
for i in range(len(signals)):
if current_position == 'SPY' and signals['spy_price'].iloc[i] == signals['52_week_low'].iloc[i]:
current_position = 'IEI'
elif current_position == 'IEI' and signals['spy_price'].iloc[i] == signals['26_week_high'].iloc[i]:
current_position = 'SPY'
positions.append(current_position)
signals['position'] = positions
return signals
def calculate_returns(signals, iei_prices):
"""Calculate strategy returns"""
# Initialize portfolio value
portfolio = pd.DataFrame(index=signals.index)
portfolio['position'] = signals['position']
# Calculate daily returns
spy_returns = signals['spy_price'].pct_change()
iei_returns = iei_prices.pct_change()
# Calculate strategy returns based on position
portfolio['daily_return'] = 0.0
for i in range(1, len(portfolio)):
if portfolio['position'].iloc[i] == 'SPY':
portfolio['daily_return'].iloc[i] = spy_returns.iloc[i]
else:
portfolio['daily_return'].iloc[i] = iei_returns.iloc[i]
# Calculate cumulative returns
portfolio['cumulative_return'] = (1 + portfolio['daily_return']).cumprod()
return portfolio
def calculate_buy_and_hold_returns(spy_prices):
"""Calculate returns for buy-and-hold SPY strategy"""
buy_hold = pd.DataFrame(index=spy_prices.index)
buy_hold['daily_return'] = spy_prices.pct_change()
buy_hold['cumulative_return'] = (1 + buy_hold['daily_return']).cumprod()
return buy_hold
def calculate_risk_metrics(portfolio):
"""Calculate risk metrics including Sharpe ratio and maximum drawdown"""
# Calculate Sharpe Ratio
risk_free_rate = 0.02 # Assuming 2% annual risk-free rate
daily_rf_rate = (1 + risk_free_rate) ** (1/252) - 1
excess_returns = portfolio['daily_return'] - daily_rf_rate
sharpe_ratio = np.sqrt(252) * (excess_returns.mean() / excess_returns.std())
# Calculate Maximum Drawdown
cum_returns = portfolio['cumulative_return']
rolling_max = cum_returns.expanding().max()
drawdowns = cum_returns / rolling_max - 1
max_drawdown = drawdowns.min()
# Calculate when max drawdown occurred
max_drawdown_idx = drawdowns.idxmin()
peak_idx = rolling_max.loc[:max_drawdown_idx].idxmax()
return {
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'max_drawdown_start': peak_idx,
'max_drawdown_end': max_drawdown_idx
}
def calculate_cagr(portfolio, start_date, end_date):
"""Calculate Compound Annual Growth Rate (CAGR)"""
total_return = portfolio['cumulative_return'].iloc[-1]
# Calculate number of years
start_date = pd.to_datetime(start_date)
end_date = pd.to_datetime(end_date)
years = (end_date - start_date).days / 365.25
# Calculate CAGR
cagr = (total_return ** (1/years)) - 1
return cagr
def main():
# Set date range
start_date = '2009-12-31'
end_date = datetime.now().strftime('%Y-%m-%d')
# Fetch historical data
spy_prices = get_historical_data('SPY', start_date, end_date)
iei_prices = get_historical_data('IEI', start_date, end_date)
# Calculate signals and returns for strategy
signals = calculate_signals(spy_prices)
portfolio = calculate_returns(signals, iei_prices)
# Calculate buy-and-hold returns
buy_hold = calculate_buy_and_hold_returns(spy_prices)
# Calculate metrics for both strategies
strategy_metrics = calculate_risk_metrics(portfolio)
buy_hold_metrics = calculate_risk_metrics(buy_hold)
strategy_cagr = calculate_cagr(portfolio, start_date, end_date)
buy_hold_cagr = calculate_cagr(buy_hold, start_date, end_date)
# Print results
initial_investment = 10000 # Example initial investment
print(f"\nBacktest Results ({start_date} to {end_date}):")
print("\nTrading Strategy Performance:")
strategy_final = initial_investment * portfolio['cumulative_return'].iloc[-1]
strategy_return = (strategy_final / initial_investment - 1) * 100
print(f"Final Value: ${strategy_final:,.2f}")
print(f"Total Return: {strategy_return:.2f}%")
print(f"CAGR: {strategy_cagr*100:.2f}%")
print(f"Sharpe Ratio: {strategy_metrics['sharpe_ratio']:.2f}")
print(f"Maximum Drawdown: {strategy_metrics['max_drawdown']*100:.2f}%")
print(f"Max Drawdown Period: {strategy_metrics['max_drawdown_start'].date()} to {strategy_metrics['max_drawdown_end'].date()}")
print("\nBuy and Hold SPY Performance:")
buy_hold_final = initial_investment * buy_hold['cumulative_return'].iloc[-1]
buy_hold_return = (buy_hold_final / initial_investment - 1) * 100
print(f"Final Value: ${buy_hold_final:,.2f}")
print(f"Total Return: {buy_hold_return:.2f}%")
print(f"CAGR: {buy_hold_cagr*100:.2f}%")
print(f"Sharpe Ratio: {buy_hold_metrics['sharpe_ratio']:.2f}")
print(f"Maximum Drawdown: {buy_hold_metrics['max_drawdown']*100:.2f}%")
print(f"Max Drawdown Period: {buy_hold_metrics['max_drawdown_start'].date()} to {buy_hold_metrics['max_drawdown_end'].date()}")
# Print trade summary
position_changes = portfolio[portfolio['position'] != portfolio['position'].shift(1)]
print(f"\nStrategy Trade Summary:")
print(f"Number of Trades: {len(position_changes)}")
print("\nTrade History:")
for date, row in position_changes.iterrows():
print(f"{date.date()}: Switch to {row['position']}")
if __name__ == "__main__":
main()
Unthreaded | Threaded | Whole Thread (8) | Prev | Next |