app.py

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
api_key_link = "https://www.alphavantage.co/support/#api-key"

# Run the Gradio app
if __name__ == "__main__":
    gr.Interface(
        fn=run_backtest,
        inputs=[
            gr.inputs.Textbox(label="API Key", placeholder=api_key_link),
            gr.inputs.Dropdown(label="From Currency", choices=from_currency_choices),
            gr.inputs.Dropdown(label="To Currency", choices=to_currency_choices),
            gr.inputs.Radio(label="Interval", choices=["1min", "5min", "15min", "30min", "60min"])
        ],
        outputs="html",  # Output as HTML to display colored text
        live=True,
        title="Trading Signal",
        description="Run Backtest and Get Trading Signal",
    ).launch()

iface = gr.Interface(fn=greet, inputs="text", outputs="text")
iface.launch()