popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

storage commeted

Browse Source
This commit is contained in:
Dobromir Popov
2024-11-18 20:12:58 +02:00
parent fee5b516a1
commit 1b0c72dc08
3 changed files with 419 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
crypto/sol/.env
View File

@ -23,18 +23,19 @@ TELEGRAM_BOT_TOKEN="6749075936:AAHUHiPTDEIu6JH7S2fQdibwsu6JVG3FNG0"
DISPLAY_CURRENCY=USD DISPLAY_CURRENCY=USD
#FOLLOW_AMOUNT=3 #FOLLOW_AMOUNT=3
FOLLOW_AMOUNT=percentage # proportional, xx%,
FOLLOW_AMOUNT=5%
LIQUIDITY_TOKENS=EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v,So11111111111111111111111111111111111111112 LIQUIDITY_TOKENS=EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v,So11111111111111111111111111111111111111112
PRIORITY=1 # 0-10, 5 = market cap, 10 twice market cap PRIORITY=1 # 0-10, 5 = market cap, 10 twice market cap
DO_WATCH_WALLET=True DO_WATCH_WALLET=True
# Niki's to Sync: [PROD] # Niki's to Sync: [PROD]
#FOLLOWED_WALLET="3EZkyU9zQRrHPnrNovDiRCA9Yg3wLK35u9cdrcGcszi1" FOLLOWED_WALLET="7TS3ATxhEVyah29gU7Z6zwwsNpWLm88ZPBig1dwyMFip"
#YOUR_WALLET="7QXGLRjvyFAmxdRaP9Wk18KwWTMfspF4Na2sr3o3PzxV" YOUR_WALLET="7QXGLRjvyFAmxdRaP9Wk18KwWTMfspF4Na2sr3o3PzxV"
#PK=3FxXjNrtEqwAKYj4BpkuLAJPzuKRWykkvjeBYQEVuFqRFWRm9eVcWrrYKbns2M31ESMoASG2WV39w9Dpx532sPUH PK=3FxXjNrtEqwAKYj4BpkuLAJPzuKRWykkvjeBYQEVuFqRFWRm9eVcWrrYKbns2M31ESMoASG2WV39w9Dpx532sPUH
# Sync to main [DEV] # Sync to main [DEV]
FOLLOWED_WALLET="7QXGLRjvyFAmxdRaP9Wk18KwWTMfspF4Na2sr3o3PzxV" #FOLLOWED_WALLET="7QXGLRjvyFAmxdRaP9Wk18KwWTMfspF4Na2sr3o3PzxV"
YOUR_WALLET="65nzyZXTLC81MthTo52a2gRJjqryTizWVqpK2fDKLye5" #YOUR_WALLET="65nzyZXTLC81MthTo52a2gRJjqryTizWVqpK2fDKLye5"
PK=5ccrMf3BFFE1HMsXt17btK1tMSNay7aBoY27saPHrqg2JEjxKBmBbxUABD9Jh7Gisf1bhM51oGzWdyLUgHdrUJPw #PK=5ccrMf3BFFE1HMsXt17btK1tMSNay7aBoY27saPHrqg2JEjxKBmBbxUABD9Jh7Gisf1bhM51oGzWdyLUgHdrUJPw

185
crypto/sol/modules/SolanaAPI.py
View File

@ -48,7 +48,7 @@ import random
import websockets import websockets
from typing import Dict, List, Optional from typing import Dict, List, Optional
import requests import requests
from datetime import datetime from datetime import datetime, timedelta
from solana.rpc.types import TokenAccountOpts, TxOpts from solana.rpc.types import TokenAccountOpts, TxOpts
from typing import List, Dict, Any, Tuple from typing import List, Dict, Any, Tuple
import traceback import traceback
@ -134,6 +134,9 @@ class SolanaWS:
elif 'error' in response_data: elif 'error' in response_data:
logger.error(f"Error in WebSocket RPC call: {response_data['error']}") logger.error(f"Error in WebSocket RPC call: {response_data['error']}")
return None return None
# if result is integer
elif "id" in response_data and int(response_data['id']) == 1:
return int(response_data['result'])
else: else:
logger.warning(f"Unexpected response: {response_data}") logger.warning(f"Unexpected response: {response_data}")
return None return None
@ -789,30 +792,36 @@ class SolanaAPI:
async def follow_move(self,move): async def follow_move(self,move):
try: try:
your_balances = await DEX.get_wallet_balances(YOUR_WALLET, doGetTokenName=False) try:
your_balance_info = next((balance for balance in your_balances.values() if balance['address'] == move['token_in']), None) your_balances = await DEX.get_wallet_balances(YOUR_WALLET, doGetTokenName=False)
if your_balance_info is not None: your_balance_info = next((balance for balance in your_balances.values() if balance['address'] == move['token_in']), None)
# Use the balance if your_balance_info is not None:
print(f"Your balance: {your_balance_info['amount']} {move['symbol_in']}") # Use the balance
else: print(f"Your balance: {your_balance_info['amount']} {move['symbol_in']}")
print(f"No ballance found for {move['symbol_in']}. Skipping move.") # else:
await telegram_utils.send_telegram_message(f"No ballance found for {move['symbol_in']}. Skipping move.") # print(f"No ballance found for {move['symbol_in']}. Skipping move.")
return # await telegram_utils.send_telegram_message(f"No ballance found for {move['symbol_in']}. Skipping move.")
# return
your_balance = your_balance_info['amount']
your_balance = your_balance_info['amount']
if not your_balance:
msg = f"<b>Move not followed:</b>\nNo balance found for token {move['symbol_in']}. Cannot follow move."
logging.warning(msg)
await telegram_utils.send_telegram_message(msg)
return
except Exception as e:
logging.error(f"Error fetching your balance: {e}")
if FOLLOW_AMOUNT == 'proportional':
return
token_info = DEX.TOKENS_INFO.get(move['token_in']) token_info = DEX.TOKENS_INFO.get(move['token_in'])
token_name_in = token_info.get('symbol') or await SAPI.get_token_metadata_symbol(move['token_in']) token_name_in = token_info.get('symbol') or await SAPI.get_token_metadata_symbol(move['token_in'])
token_name_out = DEX.TOKENS_INFO[move['token_out']].get('symbol') or await SAPI.get_token_metadata_symbol(move['token_out']) token_name_out = DEX.TOKENS_INFO[move['token_out']].get('symbol') or await SAPI.get_token_metadata_symbol(move['token_out'])
if not your_balance:
msg = f"<b>Move not followed:</b>\nNo balance found for token {move['symbol_in']}. Cannot follow move."
logging.warning(msg)
await telegram_utils.send_telegram_message(msg)
return
if FOLLOW_AMOUNT == 'percentage': if FOLLOW_AMOUNT == 'proportional':
# Calculate the amount to swap based on the same percentage as the followed move # Calculate the amount to swap based on the same percentage as the followed move
if move.get('percentage_swapped') is None: if move.get('percentage_swapped') is None:
followed_ballances = await DEX.get_wallet_balances(FOLLOWED_WALLET, doGetTokenName=False) followed_ballances = await DEX.get_wallet_balances(FOLLOWED_WALLET, doGetTokenName=False)
@ -827,11 +836,20 @@ class SolanaAPI:
amount_to_swap = 100 amount_to_swap = 100
else: else:
amount_to_swap = your_balance * (move['percentage_swapped'] / 100) amount_to_swap = your_balance * (move['percentage_swapped'] / 100)
elif FOLLOW_AMOUNT == 'exact': # if contains %, then calculate the amount to swap based on the same percentage as the followed move
amount_to_swap = move['amount_in'] elif '%' in FOLLOW_AMOUNT:
try:
percentage = float(FOLLOW_AMOUNT.strip('%'))
amount_to_swap = move['amount_in'] * (percentage / 100)
except ValueError:
msg = f"<b>Move not followed:</b>\nInvalid FOLLOW_AMOUNT '{FOLLOW_AMOUNT}'. Must be 'percentage' or a number."
logging.warning(msg)
await telegram_utils.send_telegram_message(msg)
return
else: else:
try: try:
fixed_amount = float(FOLLOW_AMOUNT) # un USD fixed_amount = float(FOLLOW_AMOUNT) # in USD
fixed_amount_in_token = fixed_amount / move["token_in_price"] fixed_amount_in_token = fixed_amount / move["token_in_price"]
amount_to_swap = min(fixed_amount_in_token, your_balance) amount_to_swap = min(fixed_amount_in_token, your_balance)
except ValueError: except ValueError:
@ -840,7 +858,7 @@ class SolanaAPI:
await telegram_utils.send_telegram_message(msg) await telegram_utils.send_telegram_message(msg)
return return
amount_to_swap = min(amount_to_swap, your_balance) # Ensure we're not trying to swap more than we have # amount_to_swap = min(amount_to_swap, your_balance) # Ensure we're not trying to swap more than we have
decimals = token_info.get('decimals') decimals = token_info.get('decimals')
# Convert to lamports # Convert to lamports
@ -848,26 +866,26 @@ class SolanaAPI:
amount_lamports = int(amount_to_swap * 10**decimals) amount_lamports = int(amount_to_swap * 10**decimals)
logging.debug(f"Calculated amount in lamports: {amount_lamports}") logging.debug(f"Calculated amount in lamports: {amount_lamports}")
if your_balance < amount_to_swap: # should not happen # if your_balance < amount_to_swap: # should not happen
msg = ( # msg = (
f"<b>Warning:</b>\n" # f"<b>Warning:</b>\n"
f"Insufficient balance: {your_balance:.6f} {token_name_in}. We want to swap {amount_to_swap:.6f}\n({move['symbol_in']}, decimals {token_info.get('decimals')} amount {amount_lamports}).\n This will probably fail. But we will try anyway." # f"Insufficient balance: {your_balance:.6f} {token_name_in}. We want to swap {amount_to_swap:.6f}\n({move['symbol_in']}, decimals {token_info.get('decimals')} amount {amount_lamports}).\n This will probably fail. But we will try anyway."
) # )
logging.warning(msg) # logging.warning(msg)
await telegram_utils.send_telegram_message(msg) # await telegram_utils.send_telegram_message(msg)
try: try:
try: # try:
notification = ( # notification = (
f"<b>Initiating move:</b>\n" # f"<b>Initiating move:</b>\n"
f"Swapping {amount_to_swap:.2f} {token_name_in} for {token_name_out}" # f"Swapping {amount_to_swap:.2f} {token_name_in} for {token_name_out}"
+ (f" ({move['percentage_swapped']:.2f}%)" if 'percentage_swapped' in move else "") # + (f" ({move['percentage_swapped']:.2f}%)" if 'percentage_swapped' in move else "")
) # )
# logging.info(notification) # # logging.info(notification)
# error_logger.info(notification) # # error_logger.info(notification)
# await telegram_utils.send_telegram_message(notification) # # await telegram_utils.send_telegram_message(notification)
except Exception as e: # except Exception as e:
logging.error(f"Error sending notification: {e}") # logging.error(f"Error sending notification: {e}")
if self.pk is None: if self.pk is None:
self.pk = await get_pk() self.pk = await get_pk()
@ -877,6 +895,7 @@ class SolanaAPI:
async_client = AsyncClient(SOLANA_WS_URL) async_client = AsyncClient(SOLANA_WS_URL)
jupiter = Jupiter(async_client, private_key) jupiter = Jupiter(async_client, private_key)
# https://station.jup.ag/api-v6/post-swap # https://station.jup.ag/api-v6/post-swap
#transaction_data = await jupiter.swap( #transaction_data = await jupiter.swap(
transaction_data = await self.swap_on_jupiter( transaction_data = await self.swap_on_jupiter(
@ -889,6 +908,7 @@ class SolanaAPI:
logging.info(f"Initiating move. Transaction data:\n {transaction_data}") logging.info(f"Initiating move. Transaction data:\n {transaction_data}")
raw_transaction = VersionedTransaction.from_bytes(base64.b64decode(transaction_data)) raw_transaction = VersionedTransaction.from_bytes(base64.b64decode(transaction_data))
# working - no priority fee # working - no priority fee
signature = private_key.sign_message(message.to_bytes_versioned(raw_transaction.message)) signature = private_key.sign_message(message.to_bytes_versioned(raw_transaction.message))
signed_txn = VersionedTransaction.populate(raw_transaction.message, [signature]) signed_txn = VersionedTransaction.populate(raw_transaction.message, [signature])
@ -896,8 +916,9 @@ class SolanaAPI:
opts = TxOpts( opts = TxOpts(
skip_preflight=False, skip_preflight=False,
preflight_commitment=Processed, preflight_commitment=Processed,
max_retries=5 # Add retries for network issues
) )
# send the transaction # send the transaction
result = await async_client.send_raw_transaction(txn=bytes(signed_txn), opts=opts) result = await async_client.send_raw_transaction(txn=bytes(signed_txn), opts=opts)
@ -913,7 +934,10 @@ class SolanaAPI:
logging.warning(f"Failed to get transaction details for {transaction_id}.\n Probably transaction failed. Retrying again...") logging.warning(f"Failed to get transaction details for {transaction_id}.\n Probably transaction failed. Retrying again...")
await asyncio.sleep(3) await asyncio.sleep(3)
except Exception as e: except Exception as e:
decoded_data = ''# base64.b64decode(transaction_data) # decode transacion data (try base58/64)
# decoded_data = base58.b58decode(transaction_data).decode('utf-8')
# decoded_data = base64.b64decode(transaction_data).decode('utf-8')
decoded_data = None
error_message = f"<b>Move Failed:</b>\n{str(e)}</b>\n{decoded_data}</b>\n{move}" error_message = f"<b>Move Failed:</b>\n{str(e)}</b>\n{decoded_data}</b>\n{move}"
logging.error(error_message) logging.error(error_message)
# log the errors to /logs/errors.log # log the errors to /logs/errors.log
@ -1045,7 +1069,8 @@ class SolanaDEX:
token_info = self.TOKENS_INFO.setdefault(token, {}) token_info = self.TOKENS_INFO.setdefault(token, {})
if 'symbol' not in token_info: if 'symbol' not in token_info:
token_info['symbol'] = await SAPI.get_token_metadata_symbol(token) token_info['symbol'] = await SAPI.get_token_metadata_symbol(token)
token_info['price'] = price token_info['price'] = price
token_info['lastUpdated'] = datetime.now().isoformat()
return prices return prices
@ -1186,6 +1211,10 @@ class SolanaDEX:
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
logging.info(f"Getting balances for wallet: {wallet_address}") logging.info(f"Getting balances for wallet: {wallet_address}")
response = None response = None
# if ballances got in last 2 minutes, return them
if "lastUpdated" in self.TOKENS_INFO and datetime.fromisoformat(self.TOKENS_INFO["lastUpdated"]) > datetime.now() - timedelta(minutes=2):
logging.info(f"Using cached balances for wallet: {wallet_address}")
return self.TOKENS_INFO
try: try:
response = await self.solana_client.get_token_accounts_by_owner_json_parsed( response = await self.solana_client.get_token_accounts_by_owner_json_parsed(
Pubkey.from_string(wallet_address), Pubkey.from_string(wallet_address),
@ -1204,46 +1233,46 @@ class SolanaDEX:
if isinstance(info, dict) and 'mint' in info and 'tokenAmount' in info: if isinstance(info, dict) and 'mint' in info and 'tokenAmount' in info:
mint = info['mint'] mint = info['mint']
decimals = int(info['tokenAmount']['decimals']) decimals = int(info['tokenAmount']['decimals'])
amount = int(info['tokenAmount']['amount']) amount = float(info['tokenAmount']['uiAmountString'])
amount = int(amount) # amount = float(info['tokenAmount']['amount'])
if amount > 1: # amount = float(amount / 10**decimals)
amount = float(amount / 10**decimals) token_name = None
if mint in self.TOKENS_INFO: if mint in self.TOKENS_INFO:
token_name = self.TOKENS_INFO[mint].get('symbol') token_name = self.TOKENS_INFO[mint].get('symbol')
elif doGetTokenName: elif doGetTokenName:
token_name = await SAPI.get_token_metadata_symbol(mint) or 'N/A' token_name = await SAPI.get_token_metadata_symbol(mint)
self.TOKENS_INFO[mint] = {'symbol': token_name} await asyncio.sleep(2)
await asyncio.sleep(2) balances[mint] = {
'name': token_name or 'N/A',
self.TOKENS_INFO[mint]['holdedAmount'] = round(amount, decimals) 'address': mint,
self.TOKENS_INFO[mint]['decimals'] = decimals 'amount': amount,
balances[mint] = { 'decimals': decimals
'name': token_name or 'N/A', }
'address': mint, self.TOKENS_INFO[mint] = {'symbol': token_name}
'amount': amount, self.TOKENS_INFO[mint] = self.TOKENS_INFO[mint].update(balances[mint])
'decimals': decimals
} try:
try: logging.debug(f"Account balance for {token_name or "N/A"} ({mint}): {amount}")
logging.debug(f"Account balance for {token_name} ({mint}): {amount}") except Exception as e:
except Exception as e: logging.error(f"Error logging account balance: {str(e)}")
logging.error(f"Error logging account balance: {str(e)}")
else: else:
logging.warning(f"Unexpected data format for account: {account}") logging.warning(f"Unexpected data format for account: {account}")
except Exception as e: except Exception as e:
logging.error(f"Error parsing account data: {str(e)}") logging.error(f"Error parsing account data: {str(e)}")
self.TOKENS_INFO["lastUpdated"] = datetime.now().isoformat()
sol_balance = await self.solana_client.get_balance(Pubkey.from_string(wallet_address))
if sol_balance.value is not None: # sol_balance = await self.solana_client.get_balance(Pubkey.from_string(wallet_address))
balances['SOL'] = { # if sol_balance.value is not None:
'name': 'SOL', # balances['SOL'] = {
'address': 'SOL', # 'name': 'SOL',
'amount': sol_balance.value / 1e9 # 'address': 'SOL',
} # 'amount': sol_balance.value / 1e9
else: # }
logging.warning(f"SOL balance response missing for wallet: {wallet_address}") # else:
# logging.warning(f"SOL balance response missing for wallet: {wallet_address}")
except Exception as e: except Exception as e:
logging.error(f"Error getting wallet balances: {str(e)}") logging.error(f"Error getting wallet balances: {str(e)} {e.error_msg}")
if response and response.value: if response and response.value:
logging.info(f"Found {len(response.value)} ({len(balances)} non zero) token accounts for wallet: {wallet_address}") logging.info(f"Found {len(response.value)} ({len(balances)} non zero) token accounts for wallet: {wallet_address}")
else: else:

532
crypto/sol/modules/webui.py
View File

@ -5,15 +5,24 @@ from concurrent.futures import ThreadPoolExecutor
import traceback import traceback
from flask import Flask, jsonify, request, render_template, redirect, url_for from flask import Flask, jsonify, request, render_template, redirect, url_for
# from flask_oauthlib.client import OAuth # from flask_oauthlib.client import OAuth
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user from flask_login import (
LoginManager,
UserMixin,
login_user,
login_required,
logout_user,
current_user,
)
import secrets import secrets
import json import json
# from crypto.sol.config import LIQUIDITY_TOKENS # from crypto.sol.config import LIQUIDITY_TOKENS
from config import LIQUIDITY_TOKENS, YOUR_WALLET from config import LIQUIDITY_TOKENS, YOUR_WALLET
from modules import storage, utils, SolanaAPI from modules import storage, utils, SolanaAPI
from modules.utils import async_safe_call, decode_instruction_data from modules.utils import async_safe_call, decode_instruction_data
from modules.storage import Storage from modules.storage import Storage
import os import os
import logging import logging
@ -21,16 +30,19 @@ from datetime import datetime
on_transaction = None on_transaction = None
def init_app(tr_handler=None): def init_app(tr_handler=None):
global on_transaction global on_transaction
on_transaction = tr_handler on_transaction = tr_handler
app = Flask(__name__, template_folder='../templates', static_folder='../static') app = Flask(__name__, template_folder="../templates", static_folder="../static")
app.secret_key = secrets.token_hex(16) app.secret_key = secrets.token_hex(16)
executor = ThreadPoolExecutor(max_workers=10) # Adjust the number of workers as needed executor = ThreadPoolExecutor(
max_workers=10
) # Adjust the number of workers as needed
login_manager = LoginManager(app) login_manager = LoginManager(app)
login_manager.login_view = 'login' login_manager.login_view = "login"
storage = Storage() storage = Storage()
# Ensure database connection # Ensure database connection
@ -55,17 +67,18 @@ def init_app(tr_handler=None):
# authorize_url='https://accounts.google.com/o/oauth2/auth', # authorize_url='https://accounts.google.com/o/oauth2/auth',
# ) # )
login_manager = LoginManager() login_manager = LoginManager()
login_manager.init_app(app) login_manager.init_app(app)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# API # API
@app.route('/tr/<wallet>/<tx_signature>', methods=['GET', 'POST']) @app.route("/tr/<wallet>/<tx_signature>", methods=["GET", "POST"])
async def transaction_notified(wallet, tx_signature): async def transaction_notified(wallet, tx_signature):
try: try:
logger.info(f"Processing transaction notification for wallet: {wallet}, tx: {tx_signature}") logger.info(
f"Processing transaction notification for wallet: {wallet}, tx: {tx_signature}"
)
request_data = request.get_json() if request.is_json else None request_data = request.get_json() if request.is_json else None
if not request_data: if not request_data:
# Process the transaction # Process the transaction
@ -73,73 +86,76 @@ def init_app(tr_handler=None):
tr = await SolanaAPI.SAPI.get_transaction_details_info(tx_signature, []) tr = await SolanaAPI.SAPI.get_transaction_details_info(tx_signature, [])
else: else:
tr = request_data tr = request_data
# ToDo - probably optimize
# ToDo - probably optimize tr["symbol_in"] = await SolanaAPI.SAPI.get_token_metadata_symbol(
tr['symbol_in'] = await SolanaAPI.SAPI.get_token_metadata_symbol(tr['token_in']) tr["token_in"]
tr['symbol_out'] = await SolanaAPI.SAPI.get_token_metadata_symbol(tr['token_out'])
prices = await SolanaAPI.DEX.get_token_prices([tr['token_in'], tr['token_out']])
tr['value_in_USD'] = prices.get(tr['token_in'], 0) * tr['amount_in']
tr['value_out_USD'] = prices.get(tr['token_out'], 0) * tr['amount_out']
notification = (
f"<b>Got TXN notification:</b>: {tr['amount_in']} {tr['symbol_in']} swapped for {tr['symbol_out']} \n"
) )
tr["symbol_out"] = await SolanaAPI.SAPI.get_token_metadata_symbol(
tr["token_out"]
)
prices = await SolanaAPI.DEX.get_token_prices(
[tr["token_in"], tr["token_out"]]
)
tr["value_in_USD"] = prices.get(tr["token_in"], 0) * tr["amount_in"]
tr["value_out_USD"] = prices.get(tr["token_out"], 0) * tr["amount_out"]
notification = f"<b>Got TXN notification:</b>: {tr['amount_in']} {tr['symbol_in']} swapped for {tr['symbol_out']} \n"
logging.info(notification) logging.info(notification)
await utils.telegram_utils.send_telegram_message(notification) await utils.telegram_utils.send_telegram_message(notification)
# Store the notified transaction in the database # Store the notified transaction in the database
original_transaction = storage.Transaction( # original_transaction = storage.Transaction(
wallet=wallet, # wallet=wallet,
transaction_type="SWAP", # transaction_type="SWAP",
symbol_in=tr['symbol_in'], # symbol_in=tr['symbol_in'],
amount_in=tr['amount_in'], # amount_in=tr['amount_in'],
value_in_usd=tr['value_in_USD'], # value_in_usd=tr['value_in_USD'],
symbol_out=tr['symbol_out'], # symbol_out=tr['symbol_out'],
amount_out=tr['amount_out'], # amount_out=tr['amount_out'],
value_out_usd=tr['value_out_USD'], # value_out_usd=tr['value_out_USD'],
tx_signature=tx_signature # tx_signature=tx_signature
) # )
await storage.store_transaction(original_transaction) # await storage.store_transaction(original_transaction)
# Attempt to execute the copytrade transaction # # Attempt to execute the copytrade transaction
try: try:
await SolanaAPI.SAPI.follow_move(tr) await SolanaAPI.SAPI.follow_move(tr)
# Store the successful copytrade transaction # Store the successful copytrade transaction
follow_transaction = storage.Transaction( # follow_transaction = storage.Transaction(
wallet=wallet, # wallet=wallet,
transaction_type="SWAP", # transaction_type="SWAP",
symbol_in=tr['symbol_in'], # symbol_in=tr['symbol_in'],
amount_in=tr['amount_in'], # amount_in=tr['amount_in'],
value_in_usd=tr['value_in_USD'], # value_in_usd=tr['value_in_USD'],
symbol_out=tr['symbol_out'], # symbol_out=tr['symbol_out'],
amount_out=tr['amount_out'], # amount_out=tr['amount_out'],
value_out_usd=tr['value_out_USD'], # value_out_usd=tr['value_out_USD'],
tx_signature=tx_signature # tx_signature=tx_signature
) # )
await storage.store_transaction(follow_transaction) # await storage.store_transaction(follow_transaction)
except Exception as e: except Exception as e:
# Store the failed copytrade transaction # # Store the failed copytrade transaction
failed_transaction = storage.Transaction( # failed_transaction = storage.Transaction(
wallet=wallet, # wallet=wallet,
transaction_type="SWAP_FAIL", # transaction_type="SWAP_FAIL",
symbol_in=tr['symbol_in'], # symbol_in=tr['symbol_in'],
amount_in=tr['amount_in'], # amount_in=tr['amount_in'],
value_in_usd=tr['value_in_USD'], # value_in_usd=tr['value_in_USD'],
symbol_out=tr['symbol_out'], # symbol_out=tr['symbol_out'],
amount_out=tr['amount_out'], # amount_out=tr['amount_out'],
value_out_usd=tr['value_out_USD'], # value_out_usd=tr['value_out_USD'],
tx_signature=tx_signature # tx_signature=tx_signature
) # )
await storage.store_transaction(failed_transaction) # await storage.store_transaction(failed_transaction)
logging.error(f"Copytrade transaction failed: {e}") logging.error(f"Copytrade transaction failed: {e}")
# ToDo - probably optimize # ToDo - probably optimize
await SolanaAPI.SAPI.save_token_info() await SolanaAPI.SAPI.save_token_info()
return jsonify(tr), 200 return jsonify(tr), 200
except Exception as e: except Exception as e:
logging.error(f"Error processing transaction: {e}") logging.error(f"Error processing transaction: {e}")
return jsonify({"error": "Failed to process transaction"}), 500 return jsonify({"error": "Failed to process transaction"}), 500
@app.route('/wh', methods=['POST']) @app.route("/wh", methods=["POST"])
async def webhook(): async def webhook():
try: try:
current_time = datetime.now().strftime("%Y%m%d-%H%M%S") current_time = datetime.now().strftime("%Y%m%d-%H%M%S")
@ -152,29 +168,30 @@ def init_app(tr_handler=None):
else: else:
logger.info(f"Webhook data: {request_data}") logger.info(f"Webhook data: {request_data}")
# save dump to /cache/last-webhook-{datetime}.json # save dump to /cache/last-webhook-{datetime}.json
with open( os.path.join(SolanaAPI.root_path, 'logs', f'wh_{current_time}.json') , 'w') as f: with open(
os.path.join(SolanaAPI.root_path, "logs", f"wh_{current_time}.json"),
"w",
) as f:
json.dump(request_data, f) json.dump(request_data, f)
if "meta" in request_data[0]: if "meta" in request_data[0]:
meta = request_data[0]["meta"] meta = request_data[0]["meta"]
# Parse inner instructions # Parse inner instructions
for inner_ix in meta.get("innerInstructions", []): for inner_ix in meta.get("innerInstructions", []):
for instruction in inner_ix.get("instructions", []): for instruction in inner_ix.get("instructions", []):
decoded = decode_instruction_data(instruction["data"]) decoded = decode_instruction_data(instruction["data"])
logger.info(f"Instruction data decoded: {decoded}") logger.info(f"Instruction data decoded: {decoded}")
# Example of pattern matching for specific instruction types # Example of pattern matching for specific instruction types
if decoded["instruction_type"] == 1: # Example: swap instruction if (
decoded["instruction_type"] == 1
): # Example: swap instruction
# Parse parameters based on program type # Parse parameters based on program type
# Different DEXes will have different parameter layouts # Different DEXes will have different parameter layouts
pass pass
# await process_wh(request_data) # await process_wh(request_data)
# don't wait for the process to finish # don't wait for the process to finish
executor.submit(asyncio.run, process_wh(request_data)) executor.submit(asyncio.run, process_wh(request_data))
@ -182,33 +199,37 @@ def init_app(tr_handler=None):
except Exception as e: except Exception as e:
logging.error(f"Error processing webhook: {e}") logging.error(f"Error processing webhook: {e}")
return jsonify({"error": "Failed to process webhook"}), 500 return jsonify({"error": "Failed to process webhook"}), 500
# Flask route to retry processing the last log # Flask route to retry processing the last log
async def process_wh(data): async def process_wh(data):
global on_transaction global on_transaction
try: try:
if data[0].get('type') == "SWAP": if data[0].get("type") == "SWAP":
swap_event = data[0]['events'].get('swap') swap_event = data[0]["events"].get("swap")
if not swap_event: if not swap_event:
logging.warning("No swap event found in data") logging.warning("No swap event found in data")
return return
# Extract token input details from the first token input # Extract token input details from the first token input
token_inputs = swap_event.get('tokenInputs', []) token_inputs = swap_event.get("tokenInputs", [])
token_outputs = swap_event.get('tokenOutputs', []) token_outputs = swap_event.get("tokenOutputs", [])
tr = {} tr = {}
wallet = data[0]['feePayer'] # Using feePayer as the wallet address wallet = data[0]["feePayer"] # Using feePayer as the wallet address
tx_signature = data[0]['signature'] tx_signature = data[0]["signature"]
usdcMint = LIQUIDITY_TOKENS[0] usdcMint = LIQUIDITY_TOKENS[0]
solMint = LIQUIDITY_TOKENS[1] solMint = LIQUIDITY_TOKENS[1]
try: try:
# Determine transaction type # Determine transaction type
if token_inputs and token_outputs and LIQUIDITY_TOKENS[0] in [token_inputs[0]["mint"], token_outputs[0]["mint"]]: if (
token_inputs
and token_outputs
and LIQUIDITY_TOKENS[0]
in [token_inputs[0]["mint"], token_outputs[0]["mint"]]
):
if token_inputs[0]["mint"] == usdcMint: if token_inputs[0]["mint"] == usdcMint:
tr["type"] = "BUY" tr["type"] = "BUY"
else: else:
@ -216,107 +237,137 @@ def init_app(tr_handler=None):
else: else:
tr["type"] = "SWAP" tr["type"] = "SWAP"
if swap_event.get("nativeInput", None):
if swap_event.get('nativeInput', None):
tr["token_in"] = solMint tr["token_in"] = solMint
tr["amount_in"] = int(swap_event.get('nativeInput')["amount"])/ 10**6 tr["amount_in"] = (
int(swap_event.get("nativeInput")["amount"]) / 10**9
)
tr["type"] = "BUY" tr["type"] = "BUY"
tr["token_in_decimals"] = 6 tr["token_in_decimals"] = 9
if swap_event.get('nativeOutput', None): if swap_event.get("nativeOutput", None):
tr["token_out"] = solMint tr["token_out"] = solMint
tr["amount_out"] = int(swap_event.get('nativeOutput')["amount"]) / 10**6 tr["amount_out"] = (
int(swap_event.get("nativeOutput")["amount"]) / 10**9
)
tr["type"] = "SELL" tr["type"] = "SELL"
tr["token_out_decimals"] = 6 tr["token_out_decimals"] = 9
if not token_inputs or len(token_inputs) == 0: # if we don't have token_in yet
logging.info("Assumed USDC as first token. BUY transaction detected") if "token_in" not in tr:
tr["token_in"] = usdcMint if not token_inputs or len(token_inputs) == 0:
tr["type"] = "BUY" logging.info(
tr["amount_in"] = await calculate_price_amount(token_outputs[0]) "Assumed USDC as first token. BUY transaction detected"
else: )
token_in = token_inputs[0] tr["token_in"] = usdcMint
tr["token_in"] = token_in["mint"] tr["type"] = "BUY"
tr["token_in_decimals"] = get_decimals(token_in) tr["amount_in"] = await calculate_price_amount(
tr["amount_in"] = calculate_amount(token_in) token_outputs[0]
)
else:
token_in = token_inputs[0]
tr["token_in"] = token_in["mint"]
tr["token_in_decimals"] = get_decimals(token_in)
tr["amount_in"] = calculate_amount(token_in)
# if we don't have token_out yet
if "token_out" not in tr:
if not token_outputs or len(token_outputs) == 0:
logging.info(
"Assumed USDC as second token. SELL transaction detected"
)
tr["token_out"] = usdcMint
tr["type"] = "SELL"
tr["amount_out"] = await calculate_price_amount(
token_inputs[0]
)
else:
token_out = token_outputs[0]
tr["token_out"] = token_out["mint"]
tr["token_out_decimals"] = get_decimals(token_out)
tr["amount_out"] = calculate_amount(token_out)
if not token_outputs or len(token_outputs) == 0:
logging.info("Assumed USDC as second token. SELL transaction detected")
tr["token_out"] = usdcMint
tr["type"] = "SELL"
tr["amount_out"] = await calculate_price_amount(token_inputs[0])
else:
token_out = token_outputs[0]
tr["token_out"] = token_out["mint"]
tr["token_out_decimals"] = get_decimals(token_out)
tr["amount_out"] = calculate_amount(token_out)
# Store transaction in database # Store transaction in database
if tr["type"] in ["BUY", "SELL"]: if tr["type"] in ["BUY", "SELL"]:
is_buy = tr["type"] == "BUY" is_buy = tr["type"] == "BUY"
transaction = storage.Transaction( # transaction = storage.Transaction(
wallet=wallet, # wallet=wallet,
transaction_type=tr["type"], # transaction_type=tr["type"],
symbol_in=tr["token_in"], # symbol_in=tr["token_in"],
amount_in=tr["amount_in"] if is_buy else 0, # amount_in=tr["amount_in"] if is_buy else 0,
value_in_usd=tr.get("swap_amount_usd", 0) if is_buy else 0, # value_in_usd=tr.get("swap_amount_usd", 0) if is_buy else 0,
symbol_out=tr["token_out"], # symbol_out=tr["token_out"],
amount_out=tr["amount_out"] if not is_buy else 0, # amount_out=tr["amount_out"] if not is_buy else 0,
value_out_usd=tr.get("swap_amount_usd", 0) if not is_buy else 0, # value_out_usd=tr.get("swap_amount_usd", 0) if not is_buy else 0,
tx_signature=tx_signature # tx_signature=tx_signature
# )
# await storage.store_transaction(transaction)
if swap_event.get("nativeInput"): # SOL
token_in = swap_event.get("nativeInput", [])
logger.info(
f"Native input (SOL) detected ({token_in["amount"]})"
)
if (
not tr["token_in"]
or not tr["token_out"]
or tr["amount_in"] == 0
or tr["amount_out"] == 0
):
logging.warning(
"Incomplete swap details found in logs. Getting details from transaction"
)
tx_signature = data[0].get("signature")
logs = data[0].get("logs", [])
tr = await SolanaAPI.SAPI.get_transaction_details_info(
tx_signature, logs
) )
await storage.store_transaction(transaction)
if swap_event.get('nativeInput'): # SOL
token_in = swap_event.get('nativeInput', [])
logger.info(f"Native input (SOL) detected ({token_in["amount"]})")
if not tr["token_in"] or not tr["token_out"] or tr["amount_in"] == 0 or tr["amount_out"] == 0:
logging.warning("Incomplete swap details found in logs. Getting details from transaction")
tx_signature = data[0].get('signature')
logs = data[0].get('logs', [])
tr = await SolanaAPI.SAPI.get_transaction_details_info(tx_signature, logs)
except Exception as e: except Exception as e:
logging.error(f"Error loading transaction token data: {str(e)}") logging.error(f"Error loading transaction token data: {str(e)}")
# ToDo - probably optimize # ToDo - probably optimize
tr['symbol_in'] = await SolanaAPI.SAPI.get_token_metadata_symbol(tr['token_in']) tr["symbol_in"] = await SolanaAPI.SAPI.get_token_metadata_symbol(
tr['symbol_out'] = await SolanaAPI.SAPI.get_token_metadata_symbol(tr['token_out']) tr["token_in"]
prices = await SolanaAPI.DEX.get_token_prices([tr['token_in'], tr['token_out']])
tr["token_in_price"] = prices.get(tr['token_in'], 0)
tr["token_out_price"] = prices.get(tr['token_out'], 0)
tr['value_in_USD'] = prices.get(tr['token_in'], 0) * tr['amount_in']
tr['value_out_USD'] = prices.get(tr['token_out'], 0) * tr['amount_out']
notification = (
f"<b>Got WH notification:</b>: {tr['amount_in']} {tr['symbol_in']} swapped for {tr['amount_out']} {tr['symbol_out']} ${tr['value_out_USD']}\n"
) )
logging.info(notification) tr["symbol_out"] = await SolanaAPI.SAPI.get_token_metadata_symbol(
await utils.telegram_utils.send_telegram_message(notification) tr["token_out"]
)
# ToDo - optimize
# prices = await SolanaAPI.DEX.get_token_prices(
# [tr["token_in"], tr["token_out"]]
# )
# tr["token_in_price"] = prices.get(tr["token_in"], 0)
# tr["token_out_price"] = prices.get(tr["token_out"], 0)
# tr["value_in_USD"] = prices.get(tr["token_in"], 0) * tr["amount_in"]
# tr["value_out_USD"] = prices.get(tr["token_out"], 0) * tr["amount_out"]
# notification = f"<b>Got WH notification:</b>: {tr['amount_in']} {tr['symbol_in'] or tr["token_in"]} swapped for {tr['amount_out']} {tr['symbol_out']} ${tr['value_out_USD']}\n"
# logging.info(notification)
# await utils.telegram_utils.send_telegram_message(notification)
# Store the notified transaction in the database # Store the notified transaction in the database
# storage.store_transaction(wallet, "SWAP", tr['symbol_in'] , tr['amount_in'], tr['value_in_USD'], tr['symbol_out'], tr['amount_out'], tr['value_out_USD'],tx_signature) # storage.store_transaction(wallet, "SWAP", tr['symbol_in'] , tr['amount_in'], tr['value_in_USD'], tr['symbol_out'], tr['amount_out'], tr['value_out_USD'],tx_signature)
copyTransaction = storage.Transaction( # copyTransaction = storage.Transaction(
wallet=wallet, # wallet=wallet,
transaction_type=tr["type"], # transaction_type=tr["type"],
symbol_in=tr["token_in"], # symbol_in=tr["token_in"],
amount_in=tr["amount_in"] if is_buy else 0, # amount_in=tr["amount_in"] if is_buy else 0,
value_in_usd=tr.get("swap_amount_usd", 0) if is_buy else 0, # value_in_usd=tr.get("swap_amount_usd", 0) if is_buy else 0,
symbol_out=tr["token_out"], # symbol_out=tr["token_out"],
amount_out=tr["amount_out"] if not is_buy else 0, # amount_out=tr["amount_out"] if not is_buy else 0,
value_out_usd=tr.get("swap_amount_usd", 0) if not is_buy else 0, # value_out_usd=tr.get("swap_amount_usd", 0) if not is_buy else 0,
tx_signature=tx_signature # tx_signature=tx_signature
) # )
try: await storage.store_transaction(copyTransaction) # try: await storage.store_transaction(copyTransaction)
except: logging.error(traceback.format_exc()) # except: logging.error(traceback.format_exc())
# Attempt to execute the copytrade transaction # Attempt to execute the copytrade transaction
try: try:
# await SolanaAPI.SAPI.follow_move(tr) # await SolanaAPI.SAPI.follow_move(tr)
if on_transaction: if on_transaction:
await async_safe_call( on_transaction, tr) await async_safe_call(on_transaction, tr)
else: else:
await SolanaAPI.SAPI.follow_move(tr) await SolanaAPI.SAPI.follow_move(tr)
# Store the successful copytrade transaction # Store the successful copytrade transaction
@ -325,7 +376,7 @@ def init_app(tr_handler=None):
# Store the failed copytrade transaction # Store the failed copytrade transaction
# await storage.store_transaction(wallet, "SWAP_FAIL", tr['symbol_in'] , tr['amount_in'], tr['value_in_USD'], tr['symbol_out'], tr['amount_out'], tr['value_out_USD'],tx_signature) # await storage.store_transaction(wallet, "SWAP_FAIL", tr['symbol_in'] , tr['amount_in'], tr['value_in_USD'], tr['symbol_out'], tr['amount_out'], tr['value_out_USD'],tx_signature)
logging.error(f"Copytrade transaction failed: {e}") logging.error(f"Copytrade transaction failed: {e}")
# ToDo - probably optimize # ToDo - probably optimize
await SolanaAPI.DEX.save_token_info() await SolanaAPI.DEX.save_token_info()
else: else:
logger.info("wh transaction is not a swap. skipping...") logger.info("wh transaction is not a swap. skipping...")
@ -335,11 +386,15 @@ def init_app(tr_handler=None):
logging.error(traceback.format_exc()) logging.error(traceback.format_exc())
def get_decimals(token_data): def get_decimals(token_data):
return token_data["rawTokenAmount"].get("decimals") or token_data["rawTokenAmount"].get("decimalFs", 0) return token_data["rawTokenAmount"].get("decimals") or token_data[
"rawTokenAmount"
].get("decimalFs", 0)
def calculate_amount(token_data): def calculate_amount(token_data):
decimals = get_decimals(token_data) decimals = get_decimals(token_data)
token_amount = int(token_data["rawTokenAmount"]["tokenAmount"]) token_amount = int(token_data["rawTokenAmount"]["tokenAmount"])
return float(token_amount / 10**decimals) return float(token_amount / 10**decimals)
async def calculate_price_amount(token_data, prices=None): async def calculate_price_amount(token_data, prices=None):
if not prices: if not prices:
prices = await SolanaAPI.DEX.get_token_prices([token_data["mint"]]) prices = await SolanaAPI.DEX.get_token_prices([token_data["mint"]])
@ -347,19 +402,19 @@ def init_app(tr_handler=None):
token_amount = int(token_data["rawTokenAmount"]["tokenAmount"]) token_amount = int(token_data["rawTokenAmount"]["tokenAmount"])
return prices[token_data["mint"]] * token_amount / 10**decimals return prices[token_data["mint"]] * token_amount / 10**decimals
@app.route('/replay_wh', methods=['POST']) @app.route("/replay_wh", methods=["POST"])
async def replay_wh(): async def replay_wh():
try: try:
data = request.get_json() data = request.get_json()
filename = data.get('filename') filename = data.get("filename")
if not filename: if not filename:
return jsonify({"error": "Filename not provided"}), 400 return jsonify({"error": "Filename not provided"}), 400
file_path = os.path.join(SolanaAPI.root_path, 'logs', filename) file_path = os.path.join(SolanaAPI.root_path, "logs", filename)
if not os.path.exists(file_path): if not os.path.exists(file_path):
return jsonify({"error": "File not found"}), 404 return jsonify({"error": "File not found"}), 404
with open(file_path, 'r') as f: with open(file_path, "r") as f:
log_data = json.load(f) log_data = json.load(f)
await process_wh(log_data) await process_wh(log_data)
@ -368,42 +423,45 @@ def init_app(tr_handler=None):
except Exception as e: except Exception as e:
logging.error(f"Error replaying webhook file: {e}") logging.error(f"Error replaying webhook file: {e}")
return jsonify({"error": "Failed to replay webhook file"}), 500 return jsonify({"error": "Failed to replay webhook file"}), 500
@app.route('/retry-last-log', methods=['GET'])
@app.route("/retry-last-log", methods=["GET"])
async def retry_last_log(): async def retry_last_log():
wh = request.args.get('wh', 'false').lower() == 'true' wh = request.args.get("wh", "false").lower() == "true"
latest_log_file = get_latest_log_file(wh) latest_log_file = get_latest_log_file(wh)
if not latest_log_file: if not latest_log_file:
return jsonify({"error": "No log files found"}), 404 return jsonify({"error": "No log files found"}), 404
try: try:
utils.log.info(f"Processing latest log file: {latest_log_file}") utils.log.info(f"Processing latest log file: {latest_log_file}")
with open(latest_log_file, 'r') as f: with open(latest_log_file, "r") as f:
log = json.load(f) log = json.load(f)
if wh: if wh:
result = await process_wh(log) result = await process_wh(log)
else: else:
result = await SolanaAPI.process_log(log) result = await SolanaAPI.process_log(log)
return jsonify({ return (
"file": latest_log_file, jsonify(
"status": "Log dump processed successfully", {
"result": result "file": latest_log_file,
}), 200 "status": "Log dump processed successfully",
"result": result,
}
),
200,
)
except Exception as e: except Exception as e:
utils.log.error(f"Error processing log dump: {e}") utils.log.error(f"Error processing log dump: {e}")
return jsonify({"error": "Failed to process log"}), 500 return jsonify({"error": "Failed to process log"}), 500
# # # #
# # # #
# AUTHENTICATION # AUTHENTICATION
# # # # # # # #
@app.route('/login/google/authorized') @app.route("/login/google/authorized")
def authorized(): def authorized():
# resp = google.authorized_response() # resp = google.authorized_response()
# if resp is None or resp.get('access_token') is None: # if resp is None or resp.get('access_token') is None:
@ -415,9 +473,8 @@ def init_app(tr_handler=None):
# user_info = google.get('userinfo') # user_info = google.get('userinfo')
# user = storage.get_or_create_user(user_info.data['email'], user_info.data['id']) # user = storage.get_or_create_user(user_info.data['email'], user_info.data['id'])
# login_user(user) # login_user(user)
return redirect(url_for('index')) return redirect(url_for("index"))
class User(UserMixin): class User(UserMixin):
def __init__(self, id, username, email): def __init__(self, id, username, email):
self.id = id self.id = id
@ -428,88 +485,107 @@ def init_app(tr_handler=None):
def load_user(user_id): def load_user(user_id):
user_data = storage.get_user_by_id(user_id) user_data = storage.get_user_by_id(user_id)
if user_data: if user_data:
return User(id=user_data['id'], username=user_data['username'], email=user_data['email']) return User(
id=user_data["id"],
username=user_data["username"],
email=user_data["email"],
)
return None return None
@app.route('/') @app.route("/")
def index(): def index():
return render_template('index.html') return render_template("index.html")
@login_manager.unauthorized_handler @login_manager.unauthorized_handler
def unauthorized(): def unauthorized():
return redirect('/login?next=' + request.path) return redirect("/login?next=" + request.path)
# return jsonify({'error': 'Unauthorized'}), 401 # return jsonify({'error': 'Unauthorized'}), 401
@app.route('/login', methods=['GET', 'POST']) @app.route("/login", methods=["GET", "POST"])
def login(): def login():
if request.method == 'POST': if request.method == "POST":
username = request.form.get('username') username = request.form.get("username")
password = request.form.get('password') password = request.form.get("password")
user = storage.authenticate_user(username, password) user = storage.authenticate_user(username, password)
if user: if user:
login_user(User(id=user['id'], username=user['username'], email=user['email'])) login_user(
return redirect(url_for('dashboard')) User(id=user["id"], username=user["username"], email=user["email"])
)
return redirect(url_for("dashboard"))
else: else:
return render_template('login.html', error='Invalid credentials') return render_template("login.html", error="Invalid credentials")
elif request.args.get('google'): elif request.args.get("google"):
# Uncomment the following line if Google OAuth is set up # Uncomment the following line if Google OAuth is set up
# return google.authorize(callback=url_for('authorized', _external=True)) # return google.authorize(callback=url_for('authorized', _external=True))
return render_template('login.html', error='Google OAuth not configured') return render_template("login.html", error="Google OAuth not configured")
return render_template('login.html') return render_template("login.html")
@app.route('/logout') @app.route("/logout")
@login_required @login_required
def logout(): def logout():
logout_user() logout_user()
return redirect(url_for('index')) return redirect(url_for("index"))
@app.route('/dashboard') @app.route("/dashboard")
@login_required @login_required
def dashboard(): def dashboard():
return render_template('dashboard.html') return render_template("dashboard.html")
@app.route('/generate_api_key', methods=['POST']) @app.route("/generate_api_key", methods=["POST"])
@login_required @login_required
def generate_api_key(): def generate_api_key():
api_key = secrets.token_urlsafe(32) api_key = secrets.token_urlsafe(32)
storage.store_api_key(current_user.id, api_key) storage.store_api_key(current_user.id, api_key)
return jsonify({'api_key': api_key}) return jsonify({"api_key": api_key})
@app.route('/wallet/<int:wallet_id>/transactions', methods=['GET']) @app.route("/wallet/<int:wallet_id>/transactions", methods=["GET"])
@login_required @login_required
@login_required @login_required
def get_transactions(wallet_id): def get_transactions(wallet_id):
transactions = storage.get_transactions(wallet_id) transactions = storage.get_transactions(wallet_id)
return jsonify(transactions) return jsonify(transactions)
@app.route('/wallet/<int:wallet_id>/holdings', methods=['GET']) @app.route("/wallet/<int:wallet_id>/holdings", methods=["GET"])
@login_required @login_required
@login_required @login_required
def get_holdings(wallet_id): def get_holdings(wallet_id):
holdings = storage.get_holdings(wallet_id) holdings = storage.get_holdings(wallet_id)
return jsonify(holdings) return jsonify(holdings)
return app return app
def teardown_app(): def teardown_app():
# Close the database connection # Close the database connection
storage.disconnect() storage.disconnect()
# Function to find the latest log file # Function to find the latest log file
def get_latest_log_file(wh:bool): def get_latest_log_file(wh: bool):
log_dir = os.path.join(SolanaAPI.root_path, 'logs') log_dir = os.path.join(SolanaAPI.root_path, "logs")
try: try:
# files = [f for f in os.listdir(log_dir) if os.path.isfile(os.path.join(log_dir, f))] # files = [f for f in os.listdir(log_dir) if os.path.isfile(os.path.join(log_dir, f))]
# filter files mask log_20241005_004103_143116.json # filter files mask log_20241005_004103_143116.json
if wh: if wh:
files = [f for f in os.listdir(log_dir) if os.path.isfile(os.path.join(log_dir, f)) and f.startswith('wh_')] files = [
f
for f in os.listdir(log_dir)
if os.path.isfile(os.path.join(log_dir, f)) and f.startswith("wh_")
]
else: else:
files = [f for f in os.listdir(log_dir) if os.path.isfile(os.path.join(log_dir, f)) and f.startswith('log_')] files = [
f
for f in os.listdir(log_dir)
if os.path.isfile(os.path.join(log_dir, f)) and f.startswith("log_")
]
latest_file = max(files, key=lambda x: os.path.getmtime(os.path.join(log_dir, x))) latest_file = max(
files, key=lambda x: os.path.getmtime(os.path.join(log_dir, x))
)
return os.path.join(log_dir, latest_file) return os.path.join(log_dir, latest_file)
except Exception as e: except Exception as e:
utils.log.error(f"Error fetching latest log file: {e}") utils.log.error(f"Error fetching latest log file: {e}")
return None return None
export = init_app, teardown_app export = init_app, teardown_app