This script reads the rsi of the lockin token from ox.fun using candles and then generates ai text and tweet based on the sentiment of the rsi and also places buy orders

Main.py

import os
import schedule
import time
import tweepy
import textwrap
from dotenv import load_dotenv
from random import randint
from openai import OpenAI

import requests
import numpy as np
import hmac
import base64
import hashlib
import datetime
import json
import logging
import threading
from tenacity import retry, wait_exponential, stop_after_attempt

# Load environment variables from .env
load_dotenv()

# OpenAI API key from .env
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# Twitter API credentials from .env
TWITTER_CONSUMER_KEY = os.getenv("TWITTER_CONSUMER_KEY")
TWITTER_CONSUMER_SECRET = os.getenv("TWITTER_CONSUMER_SECRET")
TWITTER_ACCESS_TOKEN = os.getenv("TWITTER_ACCESS_TOKEN")
TWITTER_ACCESS_TOKEN_SECRET = os.getenv("TWITTER_ACCESS_TOKEN_SECRET")

# API credentials using https://ox.fun
api_key = os.getenv("OX_API_KEY")
api_secret = os.getenv("OX_API_SECRET")
rest_url = 'https://api.ox.fun'
rest_path = 'api.ox.fun'

# Trading parameters
asset = 'LOCKIN-USD-SWAP-LIN'
rsi_period = 14
overbought = 70
oversold = 30
quantity = 10
timeframe = '60'
limit = 200
interval = 1800

# Configure logging
logging.basicConfig(level=logging.INFO)

client = OpenAI(api_key=OPENAI_API_KEY)


# Set up Twitter API authentication with tweepy.Client
twitter_client = tweepy.Client(
    consumer_key=TWITTER_CONSUMER_KEY,
    consumer_secret=TWITTER_CONSUMER_SECRET,
    access_token=TWITTER_ACCESS_TOKEN,
    access_token_secret=TWITTER_ACCESS_TOKEN_SECRET
)

# Set up Twitter API authentication with tweepy.API for media uploads
auth = tweepy.OAuth1UserHandler(
    TWITTER_CONSUMER_KEY, TWITTER_CONSUMER_SECRET, TWITTER_ACCESS_TOKEN, TWITTER_ACCESS_TOKEN_SECRET
)
twitter_api = tweepy.API(auth)

# Function to create and post a tweet
def create_tweet(tweet_message, media_id=None):
    if len(tweet_message) > 280:
        print(tweet_message)
        tweet_parts = textwrap.wrap(tweet_message, 280, break_long_words=False)
        for part in tweet_parts:
            time.sleep(randint(10, 100))
            try:
                twitter_client.create_tweet(text=part, media_ids=[media_id] if media_id else None)
            except Exception as e:
                print("An exception occurred with twitter: ", e)
                pass
    else:
        time.sleep(randint(10, 100))
        print(tweet_message)
        try:
            twitter_client.create_tweet(text=tweet_message, media_ids=[media_id] if media_id else None)
        except Exception as e:
            print("An exception occurred with twitter: ", e)
            pass

def generate_signature(api_secret, ts, nonce, verb, method, body):
    msg_string = '{}\n{}\n{}\n{}\n{}\n{}'.format(ts, nonce, verb, rest_path, method, body)
    sig = base64.b64encode(hmac.new(api_secret.encode('utf-8'), msg_string.encode('utf-8'), hashlib.sha256).digest()).decode('utf-8')
    return sig

def get_headers(api_key, api_secret, verb, method, body):
    ts = datetime.datetime.now(datetime.timezone.utc).isoformat()
    nonce = int(time.time() * 1000)
    sig = generate_signature(api_secret, ts, nonce, verb, method, body)
    headers = {
        'Content-Type': 'application/json',
        'AccessKey': api_key,
        'Timestamp': ts,
        'Signature': sig,
        'Nonce': str(nonce)
    }
    return headers

@retry(wait=wait_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(5))
def fetch_historical_data():
    method = f'v3/candles?marketCode={asset}&timeframe={timeframe}&limit={limit}'
    try:
        response = requests.get(f'{rest_url}/{method}')
        response.raise_for_status()
        data = response.json()
        if 'data' in data and isinstance(data['data'], list):
            close_prices = [float(candle['close']) for candle in data['data']]
            return close_prices
        else:
            logging.error("Error: 'data' key not found in the response or is not a list")
            logging.error(data)
            return []
    except requests.exceptions.RequestException as e:
        logging.error(f"Request error: {e}")
        raise  # Re-raise the exception to trigger retry

def calculate_rsi(prices, period):
    deltas = np.diff(prices)
    seed = deltas[:period+1]
    up = seed[seed >= 0].sum()/period
    down = -seed[seed < 0].sum()/period
    rs = up/down
    rsi = np.zeros_like(prices)
    rsi[:period] = 100. - 100./(1. + rs)

    for i in range(period, len(prices)):
        delta = deltas[i-1]
        if delta > 0:
            upval = delta
            downval = 0.
        else:
            upval = 0.
            downval = -delta

        up = (up*(period-1) + upval)/period
        down = (down*(period-1) + downval)/period
        rs = up/down
        rsi[i] = 100. - 100./(1. + rs)

    return rsi

def place_order(side, price):
    method = '/v3/orders/place'
    nonce = int(time.time() * 1000)
    body = json.dumps({
        "recvWindow": 20000,
        "responseType": "FULL",
        "timestamp": int(datetime.datetime.now().timestamp() * 1000),
        "orders": [
            {
                "clientOrderId": str(nonce),
                "marketCode": asset,
                "side": side,
                "quantity": str(quantity),
                "orderType": "MARKET",
                "price": str(price)
            }
        ]
    })
    headers = get_headers(api_key, api_secret, "POST", method, body)
    try:
        response = requests.post(rest_url + method, data=body, headers=headers)
        response.raise_for_status()
        logging.info(response.json())
        return response.json()  # Return the response to include in the tweet
    except requests.exceptions.RequestException as e:
        logging.error(f"Order placement error: {e}")
        return None

def generate_image(prompt):
    try:
        response = client.images.generate(
            prompt=prompt,
            n=1,
            size="1024x1024"
        )
        image_url = response.data[0].url
        return image_url
    except Exception as e:
        print(f"Error generating image: {e}")
        return None

def upload_image(image_url):
    try:
        response = requests.get(image_url, stream=True)
        response.raise_for_status()
        filename = 'temp_image.png'
        with open(filename, 'wb') as f:
            for chunk in response.iter_content(1024):
                f.write(chunk)
        media = twitter_api.media_upload(filename)
        os.remove(filename)
        return media.media_id_string
    except Exception as e:
        print(f"Error uploading image: {e}")
        return None

# def check_upload_status(media_id):
#     try:
#         status = twitter_api.get_media_upload_status(media_id)
#         print(f"Media upload status: {status}")
#         return status.processing_info['state'] == 'succeeded'
#     except Exception as e:
#         print(f"Error checking media upload status: {e}")
#         return False

# def add_image_metadata(media_id, alt_text):
#     try:
#         twitter_client.create_media_metadata(media_id, alt_text)
#     except Exception as e:
#         print(f"Error adding metadata to image: {e}")

def generate_and_post_tweet(message, bearish=False):
    try:
        if bearish:
            response = client.chat.completions.create(model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "You are a creative and absolutely bonkers bot that generates tweets."},
                {"role": "user", "content": (
                    "Little known fact but I’ve been abusing hard opiates for around the past 3 years. \n\n"
                    "5 days ago I built up the courage to quit. \n\n"
                    "‘For your family? For your kids?’ \n\n"
                    "No, for my future. I didn’t flush the 10s and 30s in my hand here, I sold them. "
                    "I invested all of that money back into $LOCKIN, which is a ticker I shilled to my old plug before I deleted his number. \n\n"
                    "I’ve been shaking and having cold sweats with the worst withdrawals imaginable for the past 5 days, but I’ve never felt better. \n\n"
                    "This one's for my future grandkids. $lockin"
                )},
                {"role": "user", "content": (
                    "Create an outrageous and deranged tweet starting with a similar personal story but different details, "
                    "ending with how you invested in $LOCKIN and feel better now. Include the hashtag $lockin."
                )}
            ])
            tweet_message = response.choices[0].message.content.strip() + " $lockin"
            image_prompt = "a person quitting drugs and investing in $LOCKIN, feeling better"
        else:
            response = client.chat.completions.create(model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "You are a creative and absolutely bonkers bot that generates tweets."},
                {"role": "user", "content": f"BREAKING NEWS: {message}"},
                {"role": "user", "content": "Create an outrageously funny and deranged tweet starting with BREAKING NEWS: and include the hashtag $lockin. The joke should relate to people unlocking based on the RSI and being 'locked in' if the RSI is high."}
            ])
            tweet_message = response.choices[0].message.content.strip() + " $lockin"
            image_prompt = "an outrageous scenario with people unlocking based on RSI and being locked in with $LOCKIN"

        image_url = generate_image(image_prompt)
        media_id = upload_image(image_url) if image_url else None

        # if media_id and check_upload_status(media_id):
        #     add_image_metadata(media_id, "An AI-generated image complementing the tweet's outrageous story.")
        print(f"Media ID: {media_id}")


        create_tweet(tweet_message, media_id)
    except Exception as e:
        print("Oopsie daisy! Something went bonkers while generating a hilariously deranged tweet: ", e)

def check_and_trade(): 
    try:
        close_prices = fetch_historical_data()
        if len(close_prices) > rsi_period:
            rsi = calculate_rsi(close_prices, rsi_period)[-1]
            logging.info(f"Current RSI: {rsi}")

            if rsi < oversold:
                logging.info("RSI indicates oversold, placing BUY order")
                order_response = place_order("BUY", close_prices[-1])
                if order_response:
                    tweet_message = f"Market is oversold! Bought {quantity} units of {asset} at {close_prices[-1]}. Current RSI: {rsi}."
                    generate_and_post_tweet(tweet_message)
            else:
                generate_and_post_tweet(f"Current RSI: {rsi}, not oversold. Waiting for RSI to drop below {oversold}.", bearish=True)
    except Exception as e:
        logging.error(f"Error: {e}")

def trading_bot():
    # Run the check_and_trade function immediately for testing
    check_and_trade()

    # Then continue to run it every interval
    while True:
        check_and_trade()
        time.sleep(interval)

# Start the trading bot in a separate thread
trading_thread = threading.Thread(target=trading_bot)
trading_thread.start()

while True:
    schedule.run_pending()
    time.sleep(1)  # Sleep for 1 second to avoid hitting rate limits