app.py

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'):
    try:
        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
    except Exception as e:
        print(f"An error occurred: {str(e)}")
        return None

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
    try:
        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

        # Get trade log from the strategy
        trade_log = strategy_instance.get_trade_log()

        # Iterate through the trade log and count valid trades
        valid_buy_trades = 0
        valid_sell_trades = 0

        # Initialize a variable to store the last trade line
        last_trade_line = ""

        # Iterate through the trade log and count valid trades
        for trade in trade_log:
            if trade['trade_type'] == 'CALL':
                valid_buy_trades += 1
            elif trade['trade_type'] == 'PUT':
                valid_sell_trades += 1

            # Store the last trade line
            last_trade_line = f"Trade {trade['trade_num']}: {trade['trade_type']} - {trade['outcome']}"

        for trade in trade_log:
            if trade['trade_type'] == 'CALL':
                valid_buy_trades += 1
            elif trade['trade_type'] == 'PUT':
                valid_sell_trades += 1

        # Determine if the backtest agrees (valid Buy trades > valid Sell trades)
        if win_percentage > loss_percentage:
            signal = last_trade_line
            color = "green"
        else:
            signal = last_trade_line
            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>
        """
        
        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
    except Exception as e:
        return f"Waiting for data"


# 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()