This is honestly above my knowledge I began writing it, and incorporated some chatgpt coding. I'm not that good yet. I'm just learning like most of us. Old dog new tricks.
import requests
import pandas as pd
import time
import logging
# Configure logging
logging.basicConfig(level=logging.INFO, filename='trading_bot.log',
format='%(asctime)s - %(levelname)s - %(message)s')
HIVE_API_URL = "https://api.hive.blog"
def hive_api_call(method, params):
headers = {'Content-Type': 'application/json'}
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": 1
}
response = requests.post(HIVE_API_URL, headers=headers, json=payload)
return response.json()
def get_market_history(bucket_seconds, start, end):
method = "market_history_api.get_market_history"
params = {
"bucket_seconds": bucket_seconds,
"start": start,
"end": end
}
response = hive_api_call(method, params)
return response.get("result", {}).get("buckets", [])
def get_signal(data):
data['SMA50'] = data['close'].rolling(window=50).mean()
data['SMA200'] = data['close'].rolling(window=200).mean()
if data['SMA50'].iloc[-1] > data['SMA200'].iloc[-1]:
return 'buy'
elif data['SMA50'].iloc[-1] < data['SMA200'].iloc[-1]:
return 'sell'
else:
return 'hold'
def fetch_data(symbol, bucket_seconds, start, end):
market_data = get_market_history(bucket_seconds, start, end)
df = pd.DataFrame(market_data)
df['date'] = pd.to_datetime(df['open'], unit='s')
df.set_index('date', inplace=True)
df['close'] = df['close_base'] / df['close_quote']
return df
def check_transaction_status(transaction_id, expiration=None):
method = "transaction_status_api.find_transaction"
params = {
"transaction_id": transaction_id
}
if expiration:
params["expiration"] = expiration
response = hive_api_call(method, params)
return response.get("result", {}).get("status", "unknown")
def get_account_balance(account, symbol):
method = "database_api.get_accounts"
params = {
"accounts": [account]
}
response = hive_api_call(method, params)
if 'result' in response:
balances = response['result'][0]['balances']
for balance in balances:
if balance['symbol'] == symbol:
return balance['balance']
return 0
def get_market_orders(symbol, limit=10):
method = "database_api.get_order_book"
params = {
"symbol": symbol,
"limit": limit
}
response = hive_api_call(method, params)
return response.get("result", {})
def buy(symbol, amount, account):
# Placeholder function for buying tokens
# Replace with actual buy logic or API call
logging.info(f'Buying {amount} of {symbol} for account {account}')
transaction_id = "example_buy_transaction_id" # Replace with actual transaction ID
return transaction_id
def sell(symbol, amount, account):
# Placeholder function for selling tokens
# Replace with actual sell logic or API call
logging.info(f'Selling {amount} of {symbol} for account {account}')
transaction_id = "example_sell_transaction_id" # Replace with actual transaction ID
return transaction_id
def execute_trade(signal, symbol, amount, account):
try:
if signal == 'buy':
transaction_id = buy(symbol, amount, account)
elif signal == 'sell':
transaction_id = sell(symbol, amount, account)
# Check transaction status
expiration = "2024-05-21T18:00:21" # Replace with actual expiration time if available
status = check_transaction_status(transaction_id, expiration)
logging.info(f'Transaction status for {transaction_id}: {status}')
except Exception as e:
logging.error(f'Error executing trade: {e}')
def run_bot(symbol, amount, account, bucket_seconds, interval=60):
while True:
try:
end = int(time.time())
start = end - (bucket_seconds * 200)
data = fetch_data(symbol, bucket_seconds, start, end)
signal = get_signal(data)
if signal != 'hold':
execute_trade(signal, symbol, amount, account)
time.sleep(interval)
except Exception as e:
logging.error(f'Error in run_bot loop: {e}')
time.sleep(interval)
# Run the bot
run_bot('BEE', 10, 'alice', 86400) # Example with daily buckets (86400 seconds)
