File size: 10,522 Bytes
9805f49 5d8b198 9805f49 5d8b198 bc6abba 5d8b198 54ceda9 acbf079 8c941f5 acbf079 8c941f5 acbf079 8c941f5 acbf079 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 |
import gradio as gr
import pandas as pd
import backtrader as bt
import requests
class TrendFollowingStrategy(bt.Strategy):
params = (('ma_period', 15),)
def __init__(self):
self.ma = bt.indicators.SimpleMovingAverage(self.data.close, period=self.params.ma_period)
self.crossover = bt.ind.CrossOver(self.data.close, self.ma)
self.last_signal = None
self.last_signal_timeframe = self.data._timeframe
# Additional tracking
self.trade_count = 0
self.win_count = 0
self.loss_count = 0
def next(self):
# Check if we are in the market
if not self.position:
# We are not in the market, look for a signal to enter
if self.crossover > 0:
self.buy() # Execute a buy order
self.last_signal = 'CALL'
elif self.crossover < 0:
self.sell() # Execute a sell order
self.last_signal = 'PUT'
else:
# We are in the market, look for a signal to close
if self.position.size > 0 and self.crossover < 0:
# We are long and get a sell signal
self.close() # Close the long position
elif self.position.size < 0 and self.crossover > 0:
# We are short and get a buy signal
self.close() # Close the short position
def notify_trade(self, trade):
if trade.isclosed:
outcome = 'win' if trade.pnl > 0 else 'loss'
self.log_trade(self.last_signal, outcome)
def log_trade(self, trade_type, outcome):
"""
Log the details of each trade.
"""
self.trade_count += 1
if outcome == 'win':
self.win_count += 1
elif outcome == 'loss':
self.loss_count += 1
print(f"Trade {self.trade_count}: {trade_type} - {outcome}")
def fetch_forex_intraday(api_key, from_symbol, to_symbol, interval, outputsize='compact'):
url = f'https://www.alphavantage.co/query?function=FX_INTRADAY&from_symbol={from_symbol}&to_symbol={to_symbol}&interval={interval}&apikey={api_key}&outputsize={outputsize}'
response = requests.get(url)
data = response.json()
# Extracting the time series data from the JSON object
time_series_key = 'Time Series FX (' + interval + ')'
forex_data = pd.DataFrame(data[time_series_key]).T
forex_data.columns = ['Open', 'High', 'Low', 'Close']
# Convert index to datetime and sort data
forex_data.index = pd.to_datetime(forex_data.index)
forex_data.sort_index(inplace=True)
# Convert columns to numeric
forex_data = forex_data.apply(pd.to_numeric)
return forex_data
def analyze_sentiment(json_response, target_ticker):
"""
Analyze the sentiment data for a specific ticker.
:param json_response: The JSON response from the API.
:param target_ticker: The ticker symbol to analyze (e.g., base_ticker).
:return: A string describing the overall sentiment for the target ticker.
"""
if not json_response or "feed" not in json_response:
return "No data available for analysis"
sentiment_label = "Neutral" # Default sentiment
highest_relevance = 0 # Track the highest relevance score
# Loop through each news item in the feed
for item in json_response.get("feed", []):
# Check each ticker sentiment in the item
for ticker_data in item.get("ticker_sentiment", []):
if ticker_data["ticker"] == target_ticker:
relevance_score = float(ticker_data.get("relevance_score", 0))
sentiment_score = float(ticker_data.get("ticker_sentiment_score", 0))
# Determine the sentiment label based on the score
if relevance_score > highest_relevance:
highest_relevance = relevance_score
if sentiment_score <= -0.35:
sentiment_label = "Bearish"
elif -0.35 < sentiment_score <= -0.15:
sentiment_label = "Somewhat-Bearish"
elif -0.15 < sentiment_score < 0.15:
sentiment_label = "Neutral"
elif 0.15 <= sentiment_score < 0.35:
sentiment_label = "Somewhat_Bullish"
elif sentiment_score >= 0.35:
sentiment_label = "Bullish"
return sentiment_label
def make_trade_decision(base_currency, quote_currency, quote_sentiment, base_sentiment):
"""
Make a trade decision based on sentiment analysis and forex signal parameters.
:param quote_sentiment: Sentiment analysis result for {base_currency}.
:param base_sentiment: Sentiment analysis result for {quote_currency}.
:param entry: Entry price for the trade.
:param stop_loss: Stop loss price.
:param take_profit: Take profit price.
:return: A decision to make the trade or not, along with sentiment analysis results.
"""
trade_decision = "No trade"
decision_reason = f"{base_currency} Sentiment: {quote_sentiment}, {quote_currency} Sentiment: {base_sentiment}"
# Adjust the logic to account for somewhat bullish/bearish sentiments
bullish_sentiments = ["Bullish", "Somewhat_Bullish"]
bearish_sentiments = ["Bearish", "Somewhat-Bearish"]
if quote_sentiment in bullish_sentiments and base_sentiment not in bullish_sentiments:
trade_decision = f"Sell {base_currency}/{quote_currency}"
elif base_sentiment in bullish_sentiments and quote_sentiment not in bullish_sentiments:
trade_decision = f"Buy {base_currency}/{quote_currency}"
elif quote_sentiment in bearish_sentiments and base_sentiment not in bearish_sentiments:
trade_decision = f"Buy {base_currency}/{quote_currency}"
elif base_sentiment in bearish_sentiments and quote_sentiment not in bearish_sentiments:
trade_decision = f"Sell {base_currency}/{quote_currency}"
return trade_decision, decision_reason
def fetch_sentiment_data(api_endpoint, ticker, api_key, sort='LATEST', limit=50):
# Prepare the query parameters
params = {
'function': 'NEWS_SENTIMENT',
'tickers': ticker,
'apikey': api_key,
'sort': sort,
'limit': limit
}
# Make the API request
response = requests.get(api_endpoint, params=params)
# Check if the request was successful
if response.status_code == 200:
# Return the JSON response
return response.json()
else:
# Return an error message
return f"Error fetching data: {response.status_code}"
def load_data(api_key, from_symbol, to_symbol, interval):
# Fetch data using the Alpha Vantage API
forex_data = fetch_forex_intraday(api_key, from_symbol, to_symbol, interval)
# Convert the pandas dataframe to a Backtrader data feed
data = bt.feeds.PandasData(dataname=forex_data)
return data
def should_trade(strategy, api_endpoint, api_key, base_currency, quote_currency):
consistent_periods = 3
if len(strategy) < consistent_periods:
return False, None, None, "Insufficient data"
if strategy.last_signal_timeframe in bt.TimeFrame.Names:
timeframe = bt.TimeFrame.getname(strategy.last_signal_timeframe)
else:
timeframe = "Unknown Timeframe"
base_ticker = f"FOREX:{base_currency}"
quote_ticker = f"FOREX:{quote_currency}"
# Fetch and analyze sentiment data
json_response = fetch_sentiment_data(api_endpoint, f"{base_ticker}", api_key)
print(fetch_sentiment_data(api_endpoint, f"{quote_currency}", api_key))
print(json_response)
base_sentiment = analyze_sentiment(json_response, base_ticker)
quote_sentiment = analyze_sentiment(json_response, quote_currency)
# Make a trade decision based on technical and sentiment analysis
trade_decision, decision_reason = make_trade_decision(base_currency, quote_currency, quote_sentiment, base_sentiment)
signal = strategy.crossover[0]
if all(strategy.crossover[-i] == signal for i in range(1, consistent_periods + 1)):
#timeframe = bt.TimeFrame.getname(strategy.last_signal_timeframe) if strategy.last_signal_timeframe else "Unknown Timeframe"
return True, trade_decision, timeframe, decision_reason
return False, None, None, "Not enough consistent signals or conflicting sentiment " +decision_reason+"."
# Define a function to run the backtest and provide trading signals
def run_backtest(api_key, from_symbol, to_symbol, interval):
# Set up Cerebro engine
cerebro = bt.Cerebro()
cerebro.addstrategy(TrendFollowingStrategy)
# Add data feed to Cerebro
data = load_data(api_key, from_symbol, to_symbol, interval)
cerebro.adddata(data)
# Set initial cash (optional)
cerebro.broker.set_cash(10000)
# Run the backtest
strategy_instance = cerebro.run()[0]
api_endpoint = "https://www.alphavantage.co/query" # Replace with actual endpoint
# Calculate win and loss percentages
total_trades = strategy_instance.trade_count
total_wins = strategy_instance.win_count
total_losses = strategy_instance.loss_count
win_percentage = (total_wins / total_trades) * 100
loss_percentage = (total_losses / total_trades) * 100
# Determine if it's a buy or sell based on percentages
if win_percentage > loss_percentage:
signal = "Buy"
color = "green"
else:
signal = "Sell"
color = "red"
return f"Signal: <span style='color: {color}'>{signal}</span>"
# Define a list of popular currency pairs for the dropdowns
from_currency_choices = ['EUR', 'GBP', 'USD', 'AUD', 'JPY']
to_currency_choices = ['USD', 'JPY', 'GBP', 'AUD', 'CAD']
# Placeholder link for API key
from gradio import inputs # Import the missing 'inputs' attribute from the 'gradio' module
api_key_link = "https://www.alphavantage.co/support/#api-key"
# Create a Gradio interface
gr.Interface(
fn=run_backtest,
inputs=[
gr.inputs.Textbox(label="API Key", placeholder="Enter your API key"),
gr.inputs.Dropdown(label="From Currency", choices=['EUR', 'GBP', 'USD', 'AUD', 'JPY']),
gr.inputs.Dropdown(label="To Currency", choices=['USD', 'JPY', 'GBP', 'AUD', 'CAD']),
gr.inputs.Radio(label="Interval", choices=["1min", "5min", "15min", "30min", "60min"])
],
outputs="html",
live=True,
title="Trading Signal",
description="Run Backtest and Get Trading Signal"
).launch() |