kidQuant.py

import pandas as pd
import time
from textblob import TextBlob
import ccxt

# Configuration
API_KEY = 'your_api_key'
API_SECRET = 'your_api_secret'
EXCHANGE_ID = 'binance'  # Example: Binance exchange
SYMBOL = 'BTC/USDT'  # Example trading pair
TIMEFRAME = '1m'  # Candle timeframe
DAILY_TIMEFRAME = '1d'  # Daily timeframe for trend analysis
TRADE_AMOUNT = 0.001  # Base trade amount (consider dynamic sizing)
TAKE_PROFIT_PERCENT = 1.5  # Take profit percentage
STOP_LOSS_PERCENT = 1.0  # Stop loss percentage
MOMENTUM_PERIOD = 10  # Period for trend-following logic
VOLATILITY_PERIOD = 20  # Period for volatility calculation
VOLATILITY_THRESHOLD = 0.015  # Threshold for high volatility
PAIR_SYMBOLS = ['ETH/USDT', 'BTC/USDT', 'LTC/USDT', 'BNB/USDT']  # Pairs for trading
PAIR_SPREAD_THRESHOLD = 30  # Spread threshold for pairs trading
DCA_INTERVAL = 300  # Time interval in seconds for DCA
DCA_AMOUNT = 0.001  # Amount to buy/sell in each DCA iteration
SNIPING_CONDITIONS = {
    "price_change": 5.0,  # Percentage price change for sniping
    "volume_spike": 2.0,  # Multiplier of average volume for sniping
}

# Initialize the exchange outside of conditional block
exchange = None
if CCXT_AVAILABLE:
    exchange = getattr(ccxt, EXCHANGE_ID)({
        'apiKey': API_KEY,
        'secret': API_SECRET,
        'enableRateLimit': True,
    })

# Sample DataFrame for positions
data = {
    'symbol': ['ZILUSD', 'DYDXUSD', 'LTCUSD', 'SOLUSD', 'BNBUSD', 'XRPUSD', 'ETHUSD'],
    'open_side': [None, None, None, None, None, None, 'Buy'],
    'index_pos': [0, 1, 2, 3, 4, 5, 11],
    'open_size': [0, 0, 0, 0, 0, 0, 2],
    'open_bool': [False, False, False, False, False, False, True],
    'long': [None, None, None, None, None, None, True]
}
positions_df = pd.DataFrame(data)

# Helper functions
def fetch_latest_candle(symbol, timeframe=TIMEFRAME):
    """Fetch the latest candle for the given symbol and timeframe."""
    if not exchange:
        print("Exchange is not initialized. Cannot fetch candle data.")
        return None
    candles = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=2)
    return candles[-1]  # Return the most recent candle

def calculate_daily_trend(symbol):
    """Analyze the daily trend for better-informed trading decisions."""
    if not exchange:
        print("Exchange is not initialized. Cannot calculate daily trend.")
        return None
    candles = exchange.fetch_ohlcv(symbol, timeframe=DAILY_TIMEFRAME, limit=5)  # Fetch last 5 daily candles
    if len(candles) < 5:
        return None

    # Simple trend analysis: compare the closing prices over 5 days
    closes = [c[4] for c in candles]
    trend = "up" if closes[-1] > closes[0] else "down"
    print(f"Daily trend for {symbol}: {trend}")
    return trend

def calculate_volatility(prices, period=VOLATILITY_PERIOD):
    """Calculate volatility using standard deviation."""
    if len(prices) < period:
        return None
    return pd.Series(prices).pct_change().rolling(period).std().iloc[-1]

def calculate_rsi(prices, period=14):
    """Calculate the Relative Strength Index (RSI)."""
    if len(prices) < period:
        return None

    gains = [prices[i] - prices[i - 1] for i in range(1, len(prices)) if prices[i] > prices[i - 1]]
    losses = [prices[i - 1] - prices[i] for i in range(1, len(prices)) if prices[i] < prices[i - 1]]

    avg_gain = sum(gains[-period:]) / period
    avg_loss = sum(losses[-period:]) / period

    if avg_loss == 0:
        return 100

    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

def perform_sentiment_analysis(text):
    """Perform sentiment analysis on a given text."""
    analysis = TextBlob(text)
    sentiment = analysis.sentiment.polarity
    return sentiment

def calculate_momentum(prices, period=MOMENTUM_PERIOD):
    """Calculate momentum for trend-following logic."""
    if len(prices) < period:
        return None
    return prices[-1] - prices[-period]  # Momentum as price difference

def place_order(side, amount, symbol):
    """Place a market order."""
    if not exchange:
        print("Exchange is not initialized. Cannot place order.")
        return None
    try:
        order = exchange.create_order(
            symbol=symbol,
            type='market',
            side=side,
            amount=amount
        )
        return order
    except Exception as e:
        print(f"Error placing {side} order: {e}")
        return None

def pairs_trading(symbols):
    """Execute pairs trading strategy based on price spreads."""
    prices = [fetch_latest_candle(symbol)[4] for symbol in symbols]
    if len(prices) == 2:
        spread = prices[0] - prices[1]
        print(f"Pair Spread: {spread}")
        if spread > PAIR_SPREAD_THRESHOLD:  # Arbitrary threshold
            print("Spread too wide: Short first asset, Long second asset")
            place_order('sell', TRADE_AMOUNT, symbols[0])
            place_order('buy', TRADE_AMOUNT, symbols[1])
        elif spread < -PAIR_SPREAD_THRESHOLD:
            print("Spread too negative: Long first asset, Short second asset")
            place_order('buy', TRADE_AMOUNT, symbols[0])
            place_order('sell', TRADE_AMOUNT, symbols[1])

def find_open_positions(df):
    """Find positions where open_bool is True and return the symbols."""
    open_positions = df[df['open_bool'] == True]
    symbols = open_positions['symbol'].tolist()
    print(f"Open positions found: {symbols}")
    return symbols

def token_sniping(symbols):
    """Sniping logic to monitor tokens for specific conditions."""
    for symbol in symbols:
        candle = fetch_latest_candle(symbol)
        if candle is None:
            continue
        price_change = ((candle[4] - candle[1]) / candle[1]) * 100  # % price change
        volume_spike = candle[5] / sum([c[5] for c in exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=20)])  # Volume spike ratio

        if price_change >= SNIPING_CONDITIONS["price_change"]:
            print(f"Sniping opportunity detected for {symbol}: Price change {price_change:.2f}%")
            place_order('buy', TRADE_AMOUNT, symbol)

        if volume_spike >= SNIPING_CONDITIONS["volume_spike"]:
            print(f"Volume spike detected for {symbol}: {volume_spike:.2f}x average volume")
            place_order('buy', TRADE_AMOUNT, symbol)

def dollar_cost_averaging(symbol):
    """Implement Dollar-Cost Averaging (DCA) logic."""
    print(f"Executing DCA for {symbol} with amount {DCA_AMOUNT}")
    place_order('buy', DCA_AMOUNT, symbol)

pnl_tracker = {"realized_pnl": 0.0, "unrealized_pnl": 0.0}

def update_pnl(symbol, entry_price, current_price, amount, side):
    """Update PnL tracking."""
    if side == "buy":
        pnl_tracker["unrealized_pnl"] = (current_price - entry_price) * amount
    elif side == "sell":
        pnl_tracker["realized_pnl"] += (current_price - entry_price) * amount
    print(f"PnL Update: Realized: {pnl_tracker['realized_pnl']:.2f}, Unrealized: {pnl_tracker['unrealized_pnl']:.2f}")

def trading_logic():
    """Main trading logic with enhanced features."""
    prices = []
    open_positions = find_open_positions(positions_df)

    news_headline = "Bitcoin rally continues as institutional interest surges."
    sentiment = perform_sentiment_analysis(news_headline)
    print(f"Sentiment Analysis on news headline: {sentiment}")

    daily_trend = calculate_daily_trend(SYMBOL)

    last_dca_time = time.time()

    while True:
        try:
            # Sniping logic
            token_sniping(PAIR_SYMBOLS)

            # Fetch the latest candle
            candle = fetch_latest_candle(SYMBOL)
            if candle is None:
                print("No candle data available. Skipping iteration.")
                time.sleep(5)
                continue

            close_price = candle[4]  # Closing price
            prices.append(close_price)

            # DCA logic
            if time.time() - last_dca_time >= DCA_INTERVAL:
                dollar_cost_averaging(SYMBOL)
                last_dca_time = time.time()

            # Ensure we have enough data for calculations
            if len(prices) > MOMENTUM_PERIOD:
                rsi = calculate_rsi(prices)
                momentum = calculate_momentum(prices)
                volatility = calculate_volatility(prices)

                print(f"RSI: {rsi}, Momentum: {momentum}, Volatility: {volatility}, Daily Trend: {daily_trend}")

                # Volatility arbitrage
                if volatility is not None and volatility > VOLATILITY_THRESHOLD:
                    print(f"High volatility detected ({volatility}): Placing trades.")
                    place_order('buy', TRADE_AMOUNT, SYMBOL)

                # Trend-following conditions (momentum-based)
                if momentum > 0 and sentiment > 0 and daily_trend == "up":
                    print(f"Momentum {momentum}: Buying signal with positive sentiment ({sentiment}) and upward trend.")
                    entry_price = close_price
                    place_order('buy', TRADE_AMOUNT, SYMBOL)
                    update_pnl(SYMBOL, entry_price, close_price, TRADE_AMOUNT, "buy")
                elif momentum < 0 and sentiment < 0 and daily_trend == "down":
                    print(f"Momentum {momentum}: Selling signal with negative sentiment ({sentiment}) and downward trend.")
                    entry_price = close_price
                    place_order('sell', TRADE_AMOUNT, SYMBOL)
                    update_pnl(SYMBOL, entry_price, close_price, TRADE_AMOUNT, "sell")

                # Pairs trading logic
                pairs_trading(PAIR_SYMBOLS)

            time.sleep(exchange.rateLimit / 1000 if CCXT_AVAILABLE else 5)

        except Exception as e:
            print(f"Error in trading logic: {e}")
            time.sleep(5)

if __name__ == '__main__':
    print("Starting trading bot...")
    trading_logic()