tosin2013's picture
adding more choices
acb5465
raw
history blame
12.7 kB
import gradio as gr
import pandas as pd
import backtrader as bt
import requests
# Define the Markdown-formatted instructions
with open("instructions.md", "r") as md_file:
instructions = md_file.read()
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
self.trade_log = [] # Store trade details as a list of dictionaries
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
trade_details = {
'trade_num': self.trade_count,
'trade_type': trade_type,
'outcome': outcome
}
self.trade_log.append(trade_details)
print(f"Trade {self.trade_count}: {trade_type} - {outcome}")
def get_trade_log(self):
"""
Get the trade log as a list of dictionaries.
"""
return self.trade_log
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 (' + str(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+"."
import backtrader as bt
def run_backtest(api_key, from_symbol, to_symbol, interval):
"""
Run a backtest using the specified API key, currency symbols, and interval.
Parameters:
- api_key (str): The API key for accessing the data.
- from_symbol (str): The base currency symbol.
- to_symbol (str): The quote currency symbol.
- interval (str): The time interval for the data.
Returns:
- html_message (str): An HTML message containing the calculated statistics, trade log, and trade decision information.
"""
# 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"
# Get trade decision information
trade_decision, trade_type, trade_timeframe, reason = should_trade(strategy_instance, api_endpoint, api_key, from_symbol, to_symbol)
# Create an HTML message with the calculated statistics, trade log, and trade decision information
html_message = f"""
<p><strong>Strategy Performance Summary:</strong></p>
<p>On the {interval} timeframe</p>
<p>*****************************</p>
<p>Total Trades: {total_trades}</p>
<p>Total Wins: {total_wins} ({win_percentage:.2f}%)</p>
<p>Total Losses: {total_losses} ({loss_percentage:.2f}%)</p>
<p>Signal: <span style='color: {color}'>{signal}</span></p>
<p><strong>Trade Log:</strong></p>
<ul>
"""
# Get trade log from the strategy
trade_log = strategy_instance.get_trade_log()
for trade in trade_log:
html_message += f"<li>Trade {trade['trade_num']}: {trade['trade_type']} - {trade['outcome']}</li>"
html_message += "</ul>"
# Include trade decision information
html_message += f"""
<p><strong>Trade Decision:</strong></p>
<p>Trade Type: {trade_type}</p>
<p>Timeframe: {trade_timeframe}</p>
<p>Reason: {reason}</p>
"""
return html_message
# Define a list of popular currency pairs for the dropdowns
from_currency_choices = ['EUR', 'GBP', 'USD', 'AUD', 'JPY', 'CAD', 'CHF', 'NZD']
to_currency_choices = ['USD', 'JPY', 'GBP', 'AUD', 'CAD', 'CHF', 'NZD', 'EUR']
# Placeholder link for API key
api_key_link = "https://www.alphavantage.co/support/#api-key"
api_key_input = gr.Textbox(label="API Key", placeholder="Enter your API key")
from_currency_input = gr.Dropdown(label="From Currency", choices= ['EUR', 'GBP', 'USD', 'AUD', 'JPY', 'CAD', 'CHF', 'NZD'])
to_currency_input = gr.Dropdown(label="To Currency", choices=['USD', 'JPY', 'GBP', 'AUD', 'CAD', 'CHF', 'NZD', 'EUR'])
interval_input = gr.Radio(label="Interval", choices=["1min", "5min", "15min", "30min", "60min"])
gr.Interface(
fn=run_backtest,
inputs=[api_key_input, from_currency_input, to_currency_input, interval_input],
outputs="html",
live=True,
title="Forex Trend Trading Signals",
description=instructions,
cache_examples=True
).launch()