please help me with this making sure everything is being used properly and has proper code for it use anyhow::{anyhow, Result};
use bincode::deserialize;
use bs58;
use config::Config;
use futures_util::{SinkExt, StreamExt};
use log::{debug, error, info, trace, warn};
use log4rs::{
    append::file::FileAppender,
    config::{Appender, Config as Log4rsConfig, Root},
    encode::pattern::PatternEncoder,
};
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
use log4rs::append::rolling_file::policy::compound::roll::fixed_window::FixedWindowRoller;
use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger;
use log4rs::append::rolling_file::RollingFileAppender;
use reqwest::Client;
use serde::{Deserialize, Serialize};
use serde_json::json;
use solana_client::rpc_client::RpcClient;
use solana_sdk::hash::Hash;
use solana_sdk::message::Message;
use solana_sdk::signer::keypair::Keypair;
use solana_sdk::signer::Signer;
use solana_sdk::transaction::Transaction;
use std::collections::HashMap;
use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::Mutex;
use tokio::time::interval;
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message as WsMessage};

// Configuration Constants
const WSOL_MINT: &str = "So11111111111111111111111111111111111111112";
const RAYDIUM_AMM_PROGRAM_ID: &str = "RVKd61hk3bsLCfvXGCbqfPv1dHUgAyK1fSRbmcyP2tF";
const MAX_ACTIVE_TRADES: usize = 5;
const MAX_BUY_ATTEMPTS: u32 = 3;
const BUY_ATTEMPT_DELAY: Duration = Duration::from_millis(200);
const PRICE_CHECK_INTERVAL: Duration = Duration::from_secs(1);
const STATE_FILE: &str = "bot_state.json";
const MIN_LIQUIDITY: f64 = 1000.0; // 1k SOL
const MAX_LIQUIDITY: f64 = 10000.0; // 10k SOL

#[derive(Clone, Serialize, Deserialize)]
struct Trade {
    token_address: String,
    amount: u64,
    buy_price: f64,
    take_profit_price: f64,
    stop_loss_price: f64,
    timestamp: u64,
}

struct PriceCache {
    prices: HashMap<String, (f64, Instant)>,
    ttl: Duration,
}

impl PriceCache {
    fn new(ttl: u64) -> Self {
        Self {
            prices: HashMap::new(),
            ttl: Duration::from_secs(ttl),
        }
    }

    fn get(&self, token_address: &str) -> Option<f64> {
        self.prices
            .get(token_address)
            .and_then(|(price, timestamp)| {
                if timestamp.elapsed() < self.ttl {
                    Some(*price)
                } else {
                    None
                }
            })
    }

    fn set(&mut self, token_address: String, price: f64) {
        self.prices
            .insert(token_address.clone(), (price, Instant::now()));
        debug!("Price for {} set to {}.", token_address, price);
    }

    fn clear_expired(&mut self) {
        self.prices
            .retain(|_, (_, timestamp)| timestamp.elapsed() < self.ttl);
        debug!("Cleared expired prices from the cache.");
    }
}

struct Bot {
    config: Config,
    rpc_client: RpcClient,
    keypair: Keypair,
    price_cache: Arc<Mutex<PriceCache>>,
    active_trades: Arc<Mutex<HashMap<String, Trade>>>,
    last_blockhash: Arc<Mutex<(Hash, Instant)>>,
    metadata_limiter: Arc<Mutex<()>>,
    swap_limiter: Arc<Mutex<()>>,
    http_client: Client,
}

impl Bot {
    async fn new(config_path: &str) -> Result<Self> {
        let config = Config::builder()
            .add_source(config::File::with_name(config_path))
            .build()?;

        let rpc_url = config.get_string("rpc_url")?;
        let rpc_client = RpcClient::new(rpc_url);

        let keypair = Self::load_keypair(&config)?;

        let price_cache = Arc::new(Mutex::new(PriceCache::new(60)));
        let active_trades = Arc::new(Mutex::new(HashMap::new()));
        let last_blockhash = Arc::new(Mutex::new((Hash::default(), Instant::now())));
        let metadata_limiter = Arc::new(Mutex::new(()));
        let swap_limiter = Arc::new(Mutex::new(()));
        let http_client = Client::new();

        let mut bot = Self {
            config,
            rpc_client,
            keypair,
            price_cache,
            active_trades,
            last_blockhash,
            metadata_limiter,
            swap_limiter,
            http_client,
        };

        bot.load_state().await?;
        info!("Bot initialized successfully.");
        Ok(bot)
    }

    fn load_keypair(config: &Config) -> Result<Keypair> {
        if let Ok(keypair_58) = config.get_string("keypair_58") {
            let bytes = bs58::decode(&keypair_58)
                .into_vec()
                .map_err(|e| anyhow!("Base58 decode error: {}", e))?;
            Keypair::from_bytes(&bytes)
                .map_err(|e| anyhow!("Invalid keypair bytes: {}", e))
        } else if let Ok(keypair_json_path) = config.get_string("keypair_json") {
            let json_content = std::fs::read_to_string(keypair_json_path)?;
            let secret_key: Vec<u8> = serde_json::from_str(&json_content)?;
            Keypair::from_bytes(&secret_key)
                .map_err(|e| anyhow!("Invalid JSON keypair bytes: {}", e))
        } else if let Ok(keypair_64) = config.get_string("keypair_64") {
            let bytes: Vec<u8> = keypair_64
                .split(',')
                .map(|s| s.trim().parse::<u8>().map_err(|e| anyhow!("Failed to parse byte '{}': {}", s.trim(), e)))
                .collect::<Result<Vec<u8>>>()?;
            let bytes_array: [u8; 64] = bytes[..].try_into().map_err(|_| {
                anyhow!("Keypair must be exactly 64 bytes long, but got {}", bytes.len())
            })?;
            Keypair::from_bytes(&bytes_array).map_err(|e| anyhow!("Invalid 64-byte keypair: {}", e))
        } else {
            Err(anyhow!("No valid keypair found in config"))
        }
    }

    async fn fetch_and_cache_token_price(&self, token_address: &str) -> Result<f64> {
        info!("Attempting to fetch price for token {}", token_address);

        if let Some(price) = self.price_cache.lock().await.get(token_address) {
            trace!(
                "Price for {} retrieved from cache: {}.",
                token_address,
                price
            );
            return Ok(price);
        }

        let _guard = self.metadata_limiter.lock().await;
        let price = self.fetch_token_price(token_address).await?;
        self.price_cache
            .lock()
            .await
            .set(token_address.to_string(), price);

        info!("Fetched and cached price for {}: {}", token_address, price);
        Ok(price)
    }

    async fn fetch_token_price(&self, token_address: &str) -> Result<f64> {
        debug!("Fetching price for token {}.", token_address);
        let metadata = self.fetch_metadata(token_address).await?;
        metadata
            .get("price")
            .and_then(serde_json::Value::as_f64)
            .ok_or_else(|| {
                error!("No price found for token {}", token_address);
                anyhow!("No price found for token {}", token_address)
            })
    }

    async fn fetch_metadata(&self, token_address: &str) -> Result<serde_json::Value> {
        let url = format!(
            "https://api.helius.xyz/v0/metadata/accounts?account={}",
            token_address
        );

        let response = self
            .http_client
            .get(&url)
            .header(
                "Authorization",
                format!("Bearer {}", self.config.get_string("helius_api_key")?),
            )
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        trace!("Metadata for {} fetched: {:?}", token_address, response);
        Ok(response)
    }

    async fn run(&self) -> Result<()> {
        loop {
            info!("Establishing WebSocket connection...");
            match self.connect_and_listen().await {
                Ok(_) => {
                    error!("WebSocket connection closed unexpectedly");
                }
                Err(e) => {
                    error!("Error in WebSocket connection: {}", e);
                }
            }
            info!("WebSocket connection ended. Reconnecting in 5 seconds...");
            tokio::time::sleep(Duration::from_secs(5)).await;
        }
    }

    async fn connect_and_listen(&self) -> Result<()> {
        let (ws_stream, _) =
            connect_async(&self.config.get_string("helius_websocket_url")?).await?;
        let (mut write, mut read) = ws_stream.split();

        write
            .send(WsMessage::Text(
                serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "method": "subscribePoolCreations",
                    "params": []
                })
                .to_string(),
            ))
            .await?;

        info!("Connected to Helius WebSocket. Subscribed to pool creations.");

        let mut price_check_interval = interval(PRICE_CHECK_INTERVAL);

        loop {
            tokio::select! {
                Some(msg) = read.next() => {
                    match msg {
                        Ok(WsMessage::Text(text)) => {
                            if let Err(e) = self.handle_event(&text).await {
                                error!("Error handling event: {}", e);
                            }
                        }
                        Ok(WsMessage::Close(_)) => {
                            info!("WebSocket connection closed by server");
                            return Ok(());
                        }
                        Ok(_) => {},
                        Err(e) => {
                            error!("WebSocket error: {}", e);
                            return Err(e.into());
                        }
                    }
                }
                _ = price_check_interval.tick() => {
                    if !self.active_trades.lock().await.is_empty() {
                        debug!("Performing scheduled trade checks...");
                        if let Err(e) = self.check_active_trades().await {
                            error!("Error executing scheduled trade checks: {}", e);
                        }
                    }
                }
            }
        }
    }

    async fn handle_event(&self, text: &str) -> Result<()> {
        let event: serde_json::Value = serde_json::from_str(text)?;
        if let Some(params) = event.get("params") {
            if let Some(token_address) = params.get("tokenAddress").and_then(serde_json::Value::as_str) {
                info!("Token {} detected from WebSocket event.", token_address);
                if self.is_valid_pool(token_address).await? {
                    self.attempt_buy(token_address).await?;
                } else {
                    info!("Token {} failed validation checks. Skipping.", token_address);
                }
            } else {
                warn!("No token address found in event params.");
            }
        }
        Ok(())
    }

    async fn is_valid_pool(&self, token_address: &str) -> Result<bool> {
        debug!(
            "Validating if token {} belongs to a Raydium pool.",
            token_address
        );
        let metadata = self.fetch_metadata(token_address).await?;

        let is_raydium_pool = metadata
            .get("updateAuthority")
            .and_then(serde_json::Value::as_str)
            .map_or(false, |auth| auth == RAYDIUM_AMM_PROGRAM_ID);

        if !is_raydium_pool {
            info!("Token {} is not a Raydium pool.", token_address);
            return Ok(false);
        }

        let is_spl_token = metadata.get("tokenStandard").and_then(serde_json::Value::as_str) == Some("SPL");
        if !is_spl_token {
            info!("Token {} is not an SPL standard token.", token_address);
            return Ok(false);
        }

        let liquidity = metadata
            .get("liquidity")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0);
        if liquidity < MIN_LIQUIDITY || liquidity > MAX_LIQUIDITY {
            info!("Token {} liquidity {} is out of bounds.", token_address, liquidity);
            return Ok(false);
        }

        let supply = metadata.get("supply").and_then(serde_json::Value::as_u64).unwrap_or(0);
        if supply > 1_000_000_000 {
            info!("Token {} exceeds the max supply limit.", token_address);
            return Ok(false);
        }

        if let Some(owner_count) = metadata.get("ownerCount").and_then(serde_json::Value::as_u64) {
            if owner_count > 10 {
                info!("Token {} ownership exceeds 10% limit.", token_address);
                return Ok(false);
            }
        }

        info!("Token {} passed all validation checks.", token_address);
        Ok(true)
    }

    async fn attempt_buy(&self, token_address: &str) -> Result<()> {
        let active_trades_count = self.active_trades.lock().await.len();
        if active_trades_count >= MAX_ACTIVE_TRADES {
            warn!(
                "Maximum active trades ({}) reached. Skipping buy for {}.",
                MAX_ACTIVE_TRADES, token_address
            );
            return Ok(());
        }

        let snipe_amount = self.config.get_float("snipe_amount")?;
        let snipe_amount_sol = snipe_amount * 1e9 as f64;

        for attempt in 1..=MAX_BUY_ATTEMPTS {
            info!("Buy attempt {} for token {}.", attempt, token_address);

            match self
                .execute_buy(token_address, snipe_amount_sol as u64)
                .await
            {
                Ok(_) => {
                    info!("Successfully bought token {}", token_address);
                    return Ok(());
                }
                Err(e) => {
                    warn!(
                        "Buy attempt {} failed for token {}: {}",
                        attempt, token_address, e
                    );
                    if attempt < MAX_BUY_ATTEMPTS {
                        tokio::time::sleep(BUY_ATTEMPT_DELAY).await;
                    }
                }
            }
        }

        error!("All buy attempts failed for token: {}", token_address);
        Ok(())
    }

    async fn execute_buy(&self, token_address: &str, amount: u64) -> Result<()> {
        let _guard = self.swap_limiter.lock().await;

        // Fetch Jupiter quote and swap data
        let jupiter_url = "https://quote-api.jup.ag/v6/quote";
        let params = json!({
            "inputMint": WSOL_MINT,
            "outputMint": token_address,
            "amount": amount.to_string(),
            "slippageBps": (self.config.get_float("max_slippage")? * 100.0) as u64,
        });

        debug!("Fetching quote for token {}.", token_address);
        let quote_response = self.http_client
            .post(jupiter_url)
            .query(&params)
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let swap_url = "https://quote-api.jup.ag/v6/swap";
        let swap_params = json!({
            "quoteResponse": quote_response,
            "userPublicKey": self.keypair.pubkey().to_string(),
            "wrapUnwrapSOL": true,
        });

        debug!("Preparing swap for token {}.", token_address);
        let swap_response = self.http_client
            .post(swap_url)
            .json(&swap_params)
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let transaction_data = base64::decode(swap_response["swapTransaction"].as_str().unwrap())?;
        let mut transaction: Transaction = bincode::deserialize(&transaction_data)?;

        // Sign the transaction
        let recent_blockhash = self.get_recent_blockhash().await?;
        transaction.sign(&[&self.keypair], recent_blockhash);

        // Serialize the signed transaction
        let serialized_transaction = bincode::serialize(&transaction)?;

        // Use Helius API to send the transaction with MEV protection
        let helius_url = format!("{}/v0/transactions", self.config.get_string("helius_rpc_url")?);
        let helius_params = json!({
            "transaction": base64::encode(&serialized_transaction),
        });

        debug!("Sending transaction via Helius API for MEV protection");
        let helius_response = self.http_client
            .post(&helius_url)
            .header("Authorization", format!("Bearer {}", self.config.get_string("helius_api_key")?))
            .json(&helius_params)
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let signature = helius_response["signature"].as_str()
            .ok_or_else(|| anyhow!("No signature in Helius response"))?;

        info!("Buy executed successfully with Helius MEV protection. Signature: {}", signature);

        // Add trade to active trades
        let current_price = self.fetch_and_cache_token_price(token_address).await?;
        let take_profit_price = current_price * (1.0 + self.config.get_float("profit_target_percent")? / 100.0);
        let stop_loss_price = current_price * (1.0 - self.config.get_float("stop_loss_percent")? / 100.0);

        let trade = Trade {
            token_address: token_address.to_string(),
            amount,
            buy_price: current_price,
            take_profit_price,
            stop_loss_price,
            timestamp: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("Time went backwards")
                .as_secs(),
        };

        self.active_trades
            .lock()
            .await
            .insert(token_address.to_string(), trade);

        info!(
            "Added trade to active trades. Buy price: {}, Take profit: {}, Stop loss: {}",
            current_price, take_profit_price, stop_loss_price
        );

        Ok(())
    }

    async fn check_active_trades(&self) -> Result<()> {
        let mut trades_to_remove = Vec::new();
        let active_trades = self.active_trades.lock().await;

        for (token_address, trade) in active_trades.iter() {
            let current_price = self.fetch_and_cache_token_price(token_address).await?;
            info!("Current price for {}: {}", token_address, current_price);

            if current_price >= trade.take_profit_price {
                info!(
                    "Executing sell for token {}. Profit target reached at {}.",
                    token_address, current_price
                );
                if let Err(e) = self.execute_sell(token_address).await {
                    error!("Failed to sell token {}: {}", token_address, e);
                } else {
                    trades_to_remove.push(token_address.clone());
                }
            } else if current_price <= trade.stop_loss_price {
                info!(
                    "Executing sell for token {}. Stop loss triggered at {}.",
                    token_address, current_price
                );
                if let Err(e) = self.execute_sell(token_address).await {
                    error!("Failed to sell token {}: {}", token_address, e);
                } else {
                    trades_to_remove.push(token_address.clone());
                }
            }
        }

        if !trades_to_remove.is_empty() {
            let mut active_trades = self.active_trades.lock().await;
            for token_address in trades_to_remove {
                active_trades.remove(&token_address);
                info!(
                    "Removed trade for token {} after successful sell.",
                    token_address
                );
            }
        }

        Ok(())
    }

    async fn execute_sell(&self, token_address: &str) -> Result<()> {
        let _guard = self.swap_limiter.lock().await;

        let amount = {
            let active_trades = self.active_trades.lock().await;
            active_trades
                .get(token_address)
                .map(|trade| trade.amount)
                .ok_or_else(|| {
                    error!("Trade not found for token {}", token_address);
                    anyhow!("Trade not found for token {}", token_address)
                })?
        };

        // Fetch Jupiter quote and swap data
        let jupiter_url = "https://quote-api.jup.ag/v6/quote";
        let params = json!({
            "inputMint": token_address,
            "outputMint": WSOL_MINT,
            "amount": amount.to_string(),
            "slippageBps": (self.config.get_float("max_slippage")? * 100.0) as u64,
        });

        debug!("Fetching quote for selling token {}.", token_address);
        let quote_response = self.http_client
            .post(jupiter_url)
            .query(&params)
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let swap_url = "https://quote-api.jup.ag/v6/swap";
        let swap_params = json!({
            "quoteResponse": quote_response,
            "userPublicKey": self.keypair.pubkey().to_string(),
            "wrapUnwrapSOL": true,
        });

        debug!("Preparing swap for selling token {}.", token_address);
        let swap_response = self.http_client
            .post(swap_url)
            .json(&swap_params)
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let transaction_data = base64::decode(swap_response["swapTransaction"].as_str().unwrap())?;
        let mut transaction: Transaction = bincode::deserialize(&transaction_data)?;

        // Sign the transaction
        let recent_blockhash = self.get_recent_blockhash().await?;
        transaction.sign(&[&self.keypair], recent_blockhash);

        // Serialize the signed transaction
        let serialized_transaction = bincode::serialize(&transaction)?;

        // Use Helius API to send the transaction with MEV protection
        let helius_url = format!("{}/v0/transactions", self.config.get_string("helius_rpc_url")?);
        let helius_params = json!({
            "transaction": base64::encode(&serialized_transaction),
        });

        debug!("Sending sell transaction via Helius API for MEV protection");
        let helius_response = self.http_client
            .post(&helius_url)
            .header("Authorization", format!("Bearer {}", self.config.get_string("helius_api_key")?))
            .json(&helius_params)
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let signature = helius_response["signature"].as_str()
            .ok_or_else(|| anyhow!("No signature in Helius response"))?;

        info!("Sell executed successfully with Helius MEV protection. Signature: {}", signature);

        Ok(())
    }

    async fn get_recent_blockhash(&self) -> Result<Hash> {
        let last_blockhash = self.last_blockhash.lock().await;
        if last_blockhash.1.elapsed() > Duration::from_secs(60) {
            drop(last_blockhash);
            let new_blockhash = self.rpc_client.get_latest_blockhash()?;
            let mut last_blockhash = self.last_blockhash.lock().await;
            *last_blockhash = (new_blockhash, Instant::now());
            debug!("Updated blockhash to {}", new_blockhash);
            Ok(new_blockhash)
        } else {
            Ok(last_blockhash.0)
        }
    }

    async fn save_state(&self) -> Result<()> {
        let active_trades = self.active_trades.lock().await;
        let serialized = serde_json::to_string(&*active_trades)?;
        let mut file = OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(STATE_FILE)?;
        file.write_all(serialized.as_bytes())?;
        info!("State saved to {}.", STATE_FILE);
        Ok(())
    }

    async fn load_state(&mut self) -> Result<()> {
        let mut file = match File::open(STATE_FILE) {
            Ok(file) => file,
            Err(_) => {
                warn!(
                    "State file not found: {}. Starting with an empty state.",
                    STATE_FILE
                );
                return Ok(());
            }
        };

        let mut contents = String::new();
        file.read_to_string(&mut contents)?;
        let loaded_trades: HashMap<String, Trade> = serde_json::from_str(&contents)?;
        let mut active_trades = self.active_trades.lock().await;
        *active_trades = loaded_trades;
        info!("State loaded from {}.", STATE_FILE);
        Ok(())
    }
}

fn setup_logging() -> Result<()> {
    let log_file_path = "bot.log";
    let archived_log_pattern = "bot.{}.log";
    
    let window_size = 2; // Number of archived log files to keep
    let max_size = 1 * 1024 * 1024; // 1MB

    let fixed_window_roller = FixedWindowRoller::builder()
        .build(archived_log_pattern, window_size)
        .unwrap();

    let size_trigger = SizeTrigger::new(max_size);

    let compound_policy = CompoundPolicy::new(
        Box::new(size_trigger),
        Box::new(fixed_window_roller),
    );

    let rolling_appender = RollingFileAppender::builder()
        .encoder(Box::new(PatternEncoder::new("{d(%Y-%m-%d %H:%M:%S)} - {l} - {m}{n}")))
        .build(log_file_path, Box::new(compound_policy))
        .unwrap();

    let config = Log4rsConfig::builder()
        .appender(Appender::builder().build("rolling", Box::new(rolling_appender)))
        .build(Root::builder().appender("rolling").build(log::LevelFilter::Info))
        .unwrap();

    log4rs::init_config(config)?;

    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    setup_logging()?;
    info!("Logging initialized. Starting bot...");

    let bot = Bot::new("config.toml").await?;

    let (tx, rx) = tokio::sync::oneshot::channel();
    let tx = Arc::new(StdMutex::new(Some(tx)));
    let tx_clone = tx.clone();

    ctrlc::set_handler(move || {
        if let Some(tx) = tx_clone.lock().unwrap().take() {
            let _ = tx.send(());
        }
    })
    .expect("Error setting Ctrl+C handler");

    tokio::select! {
        result = bot.run() => {
            if let Err(e) = result {
                error!("Bot error: {}", e);
            }
        }
        _ = rx => {
            info!("Received shutdown signal. Saving state and exiting...");
            if let Err(e) = bot.save_state().await {
                error!("Error saving state on shutdown: {}", e);
            }
        }
    }

    Ok(())
}