can you please fix this use pyo3::prelude::*;
use tokio::sync::{broadcast, Mutex, mpsc};
use tokio_tungstenite::connect_async;
use serde_json::{json, Value};
use serde::{Serialize, Deserialize};
use std::{sync::Arc, time::Duration, fs};
use anyhow::Result;
use tokio::time::sleep;
use solana_sdk::signer::keypair::Keypair;
use solana_client::rpc_client::RpcClient;
use rayon::prelude::*;

const RAYDIUM_PROGRAM_ID: &str = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8";
const JUPITER_V6: &str = "https://quote-api.jup.ag/v6";
const MAX_RETRIES: u8 = 3;
const MIN_LIQUIDITY: f64 = 50_000.0;
const PRIORITY_MICRO_LAMPORTS: u64 = 15000;
const MAX_CONCURRENT_TRADES: usize = 10;

#[derive(Debug, Clone, Default)]
struct TradeConfig {
    max_position_size: f64,
    risk_per_trade: f64,
    min_profit_target: f64,
    max_slippage: f64,
    emergency_stop_loss: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Config {
    network: NetworkConfig,
    sniping: SnipingConfig,
    scalping: ScalpingConfig,
    pools: Vec<PoolConfig>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct NetworkConfig {
    api_key: String,
    rpc_url: String,
    ws_url: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct SnipingConfig {
    buy_amount: f64,
    max_slippage: u64,
    priority_fee: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct ScalpingConfig {
    position_size: f64,
    take_profit: f64,
    stop_loss: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct PoolConfig {
    token: String,
    raydium_pool: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Position {
    token: String,
    entry_price: f64,
    size: f64,
    timestamp: i64,
    trade_start: i64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct JupiterQuote {
    swap_data: String,
    price_impact: f64,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
struct MarketData {
    token: String,
    price: f64,
    volume_5m: f64,
    volume_1h: f64,
    liquidity: f64,
    buy_pressure: f64,
    volatility: f64,
    timestamp: i64,
    stop_loss: f64,
    take_profit: f64,
}

struct ConnectionManager {
    client: reqwest::Client,
    ws_stream: Option<tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>>,
    rpc_client: RpcClient,
    config: Config,
}

impl ConnectionManager {
    fn new() -> Self {
        let config: Config = serde_json::from_str(
            &fs::read_to_string("config.json").expect("Failed to read config")
        ).expect("Failed to parse config");

        Self {
            client: reqwest::Client::builder()
                .timeout(Duration::from_secs(5))
                .pool_max_idle_per_host(100)
                .build()
                .unwrap(),
            ws_stream: None,
            rpc_client: RpcClient::new(config.network.rpc_url.clone()),
            config,
        }
    }

    async fn connect(&mut self) -> Result<()> {
        let ws_url = &self.config.network.ws_url;
        let (ws_stream, _) = connect_async(ws_url).await?;
        self.ws_stream = Some(ws_stream);
        Ok(())
    }
}

struct TurboEngine {
    connection: Arc<Mutex<ConnectionManager>>,
    market_data: Arc<Mutex<Vec<MarketData>>>,
    trade_config: TradeConfig,
    positions: Arc<Mutex<Vec<Position>>>,
    trade_semaphore: Arc<tokio::sync::Semaphore>,
    torch: PyObject,
    device: PyObject,
}

impl TurboEngine {
    async fn new(py: Python<'_>) -> PyResult<Self> {
        let torch = PyModule::import(py, "torch")?;
        let device = if torch.getattr("cuda")?.getattr("is_available")?.call0()?.extract()? {
            torch.getattr("device")?.call1(("cuda",))?.into()
        } else {
            torch.getattr("device")?.call1(("cpu",))?.into()
        };

        let conn_manager = ConnectionManager::new();
        let config = conn_manager.config.clone();

        Ok(Self {
            connection: Arc::new(Mutex::new(conn_manager)),
            market_data: Arc::new(Mutex::new(Vec::new())),
            trade_config: TradeConfig {
                max_position_size: config.scalping.position_size,
                risk_per_trade: 0.02,
                min_profit_target: config.scalping.take_profit,
                max_slippage: config.sniping.max_slippage as f64 / 100.0,
                emergency_stop_loss: config.scalping.stop_loss,
            },
            positions: Arc::new(Mutex::new(Vec::new())),
            trade_semaphore: Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_TRADES)),
            torch: torch.into(),
            device,
        })
    }

    fn analyze_market_torch(&self, py: Python<'_>, data: &[f64]) -> PyResult<MarketData> {
        let torch = self.torch.as_ref(py);
        let tensor = torch.call_method1("tensor", (data.to_vec(),))?;
        let cuda_tensor = tensor.call_method1("to", (&self.device,))?;
        
        let price: f64 = cuda_tensor.call_method1("get", (0_i64,))?.call_method0("item")?.extract()?;
        let volume: f64 = cuda_tensor.call_method1("get", (1_i64,))?.call_method0("item")?.extract()?;
        
        Ok(MarketData {
            token: "".to_string(),
            price,
            volume_5m: volume / 288.0,
            volume_1h: volume / 24.0,
            liquidity: price * volume,
            buy_pressure: 0.0,
            volatility: 0.0,
            timestamp: chrono::Utc::now().timestamp(),
            stop_loss: price * self.trade_config.emergency_stop_loss,
            take_profit: price * (1.0 + self.trade_config.min_profit_target),
        })
    }

    async fn execute_priority_trade(&self, quote: &JupiterQuote, keypair: &Keypair) -> Result<String> {
        let conn = self.connection.lock().await;
        let tx_data = json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "sendTransaction",
            "params": [
                quote.swap_data,
                {
                    "encoding": "base64",
                    "maxRetries": MAX_RETRIES,
                    "skipPreflight": true,
                    "priorityFee": PRIORITY_MICRO_LAMPORTS
                }
            ]
        });

        let signed_data = keypair.sign_message(&tx_data.to_string().as_bytes());
        
        let response = conn.client.post(&conn.config.network.rpc_url)
            .json(&tx_data)
            .header("Authorization", &conn.config.network.api_key)
            .send()
            .await?
            .json::<Value>()
            .await?;

        Ok(response["result"].as_str().unwrap_or_default().to_string())
    }

    async fn monitor_positions(&self) {
        let (tx, mut rx) = mpsc::channel::<MarketData>(100);
        let positions = self.positions.clone();
        
        tokio::spawn(async move {
            while let Some(market_data) = rx.recv().await {
                let positions = positions.lock().await;
                for position in positions.iter() {
                    if market_data.price >= position.entry_price * (1.0 + self.trade_config.min_profit_target) {
                        self.close_position(&position, "take_profit").await.ok();
                    } else if market_data.price <= position.entry_price * self.trade_config.emergency_stop_loss {
                        self.close_position(&position, "stop_loss").await.ok();
                    }
                }
            }
        });
    }
}

#[pymethods]
impl TurboPyTrader {
    #[new]
    fn new() -> PyResult<Self> {
        init_pytorch_env()?;
        
        let (tx, _) = broadcast::channel(1000);
        Python::with_gil(|py| {
            let engine = futures::executor::block_on(TurboEngine::new(py))
                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
            
            Ok(Self {
                engine: Arc::new(engine),
                trade_tx: tx,
                active: false,
            })
        })
    }

    fn start<'p>(&mut self, py: Python<'p>) -> PyResult<&'p PyAny> {
        self.active = true;
        let engine = self.engine.clone();
        let trade_tx = self.trade_tx.clone();

        pyo3_asyncio::tokio::future_into_py(py, async move {
            println!("🚀 TurboTrader Active - CUDA Enabled");

            engine.monitor_positions().await;

            while let Ok(market_data) = engine.process_pool_parallel().await {
                for data in market_data {
                    if data.liquidity >= MIN_LIQUIDITY {
                        if let Ok(signature) = engine.execute_trade(&data).await {
                            let sig_clone = signature.clone();
                            trade_tx.send(signature).unwrap_or_default();
                            println!("✅ Trade executed: {}", sig_clone);
                        }
                    }
                }
                sleep(Duration::from_millis(100)).await;
            }
            Ok(())
        })
    }


    async fn load_keypair(&self) -> Result<Keypair> {
        let keypair_bytes = fs::read("C:\\Users\\Orsen\\anaconda3\\envs\\pytorch_cuda\\keypair.bin")?;
        Ok(Keypair::from_bytes(&keypair_bytes)?)
    }

    async fn get_jupiter_quote(&self, token: &str) -> Result<JupiterQuote> {
        let conn = self.connection.lock().await;
        let quote_request = serde_json::json!({
            "inputMint": token,
            "amount": self.trade_config.max_position_size,
            "slippageBps": (self.trade_config.max_slippage * 10000.0) as u64,
            "priorityFee": PRIORITY_MICRO_LAMPORTS,
        });

        let response = conn.client.post(JUPITER_V6)
            .json(&quote_request)
            .send()
            .await?
            .json::<Value>()
            .await?;

        Ok(JupiterQuote {
            swap_data: response["swapTransaction"].as_str().unwrap_or_default().to_string(),
            price_impact: response["priceImpactPct"].as_f64().unwrap_or_default(),
        })
    }

    fn analyze_market_torch(&self, py: Python<'_>, data: Vec<f64>) -> PyResult<MarketData> {
        let torch = self.torch.as_ref(py);
        let tensor = torch.call_method1("tensor", (data,))?;
        let cuda_tensor = tensor.call_method1("to", (&self.device,))?;
        
        let price = cuda_tensor.call_method1("get", (0,))?.call_method0("item")?.extract()?;
        let volume = cuda_tensor.call_method1("get", (1,))?.call_method0("item")?.extract()?;
        
        Ok(MarketData {
            token: "".to_string(),
            price,
            volume_5m: volume / 288.0,
            volume_1h: volume / 24.0,
            liquidity: price * volume,
            buy_pressure: 0.0,
            volatility: 0.0,
            timestamp: chrono::Utc::now().timestamp(),
            stop_loss: price * self.trade_config.emergency_stop_loss,
            take_profit: price * (1.0 + self.trade_config.min_profit_target),
        })
    }

    async fn execute_trade(&self, market_data: &MarketData) -> Result<String> {
        let _permit = self.trade_semaphore.clone().acquire_owned().await?;
        
        let keypair = self.load_keypair().await?;
        let quote = self.get_jupiter_quote(&market_data.token).await?;
        
        if quote.price_impact > self.trade_config.max_slippage {
            return Err(anyhow::anyhow!("Price impact too high"));
        }
        
        let signature = self.execute_priority_trade(&quote, &keypair).await?;
        
        let position = Position {
            token: market_data.token.clone(),
            entry_price: market_data.price,
            size: self.trade_config.max_position_size,
            timestamp: chrono::Utc::now().timestamp(),
            trade_start: chrono::Utc::now().timestamp(),
        };
        
        self.positions.lock().await.push(position.clone());
        
        Ok(signature)
    }

    async fn execute_priority_trade(&self, quote: &JupiterQuote, keypair: &Keypair) -> Result<String> {
        let conn = self.connection.lock().await;
        let tx_data = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "sendTransaction",
            "params": [
                quote.swap_data,
                {
                    "encoding": "base64",
                    "maxRetries": MAX_RETRIES,
                    "skipPreflight": true,
                    "priorityFee": PRIORITY_MICRO_LAMPORTS
                }
            ]
        });

        let response = conn.client.post(&conn.config.network.rpc_url)
            .json(&tx_data)
            .send()
            .await?
            .json::<Value>()
            .await?;

        Ok(response["result"].as_str().unwrap_or_default().to_string())
    }

    async fn process_pool_parallel(&self) -> Result<Vec<MarketData>> {
        let conn = self.connection.lock().await;
        let response = conn.client.get(&format!("{}/v2/pools", &conn.config.network.rpc_url))
            .header("Authorization", &conn.config.network.api_key)
            .send()
            .await?
            .json::<Value>()
            .await?;

        let pools = response["pools"].as_array()
            .ok_or_else(|| anyhow::anyhow!("Invalid pool data"))?;

        let market_data: Vec<MarketData> = pools.par_iter()
            .filter(|pool| pool["address"] == RAYDIUM_PROGRAM_ID)
            .filter_map(|pool| {
                Python::with_gil(|py| self.analyze_market_torch(py, vec![
                    pool["price"].as_f64()?,
                    pool["volume"].as_f64()?
                ]).ok())
            })
            .collect();

        Ok(market_data)
    }

    async fn monitor_positions(&self) {
        let (tx, mut rx) = mpsc::channel(100);
        let positions = self.positions.clone();
        
        tokio::spawn(async move {
            while let Some(position) = rx.recv().await {
                if let Ok(market_data) = self.get_market_data(&position.token).await {
                    if market_data.price >= position.entry_price * (1.0 + self.trade_config.min_profit_target) {
                        self.close_position(&position, "take_profit").await.ok();
                    } else if market_data.price <= position.entry_price * self.trade_config.emergency_stop_loss {
                        self.close_position(&position, "stop_loss").await.ok();
                    }
                }
            }
        });

        loop {
            let positions = positions.lock().await.clone();
            for position in positions {
                tx.send(position).await.ok();
            }
            sleep(Duration::from_secs(1)).await;
        }
    }

    async fn get_market_data(&self, token: &str) -> Result<MarketData> {
        let conn = self.connection.lock().await;
        let response = conn.client.get(&format!("{}/v2/prices", &conn.config.network.rpc_url))
            .header("Authorization", &conn.config.network.api_key)
            .send()
            .await?
            .json::<Value>()
            .await?;

        let price = response["price"].as_f64().unwrap_or_default();

        Ok(MarketData {
            token: token.to_string(),
            price,
            volume_5m: response["volume5m"].as_f64().unwrap_or_default(),
            volume_1h: response["volume1h"].as_f64().unwrap_or_default(),
            liquidity: response["liquidity"].as_f64().unwrap_or_default(),
            buy_pressure: response["buyPressure"].as_f64().unwrap_or_default(),
            volatility: response["volatility"].as_f64().unwrap_or_default(),
            timestamp: chrono::Utc::now().timestamp(),
            stop_loss: price * self.trade_config.emergency_stop_loss,
            take_profit: price * (1.0 + self.trade_config.min_profit_target),
        })
    }

    async fn close_position(&self, position: &Position, reason: &str) -> Result<String> {
        let keypair = self.load_keypair().await?;
        let quote = self.get_jupiter_quote(&position.token).await?;
        let signature = self.execute_priority_trade(&quote, &keypair).await?;
        
        let mut positions = self.positions.lock().await;
        positions.retain(|p| p.token != position.token);
        
        println!("Closed position {} - Reason: {}", position.token, reason);
        Ok(signature)
    }
}

#[pyclass]
struct TurboPyTrader {
    engine: Arc<TurboEngine>,
    trade_tx: broadcast::Sender<String>,
    active: bool,
}

fn init_pytorch_env() -> PyResult<()> {
    let cuda_path = r"C:\Users\Orsen\anaconda3\envs\pytorch_cuda";
    env::set_var("PYTHONHOME", cuda_path);
    env::set_var("PYTHONPATH", format!("{}/Lib/site-packages", cuda_path));
    env::set_var("PYO3_PYTHON", format!("{}/python.exe", cuda_path));
    Ok(())
}

#[pymethods]
impl TurboPyTrader {
    #[new]
    fn new() -> PyResult<Self> {
        init_pytorch_env()?;
        
        let (tx, _) = broadcast::channel(1000);
        Python::with_gil(|py| {
            let engine = match futures::executor::block_on(TurboEngine::new(py)) {
                Ok(engine) => engine,
                Err(e) => return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))
            };
            
            Ok(Self {
                engine: Arc::new(engine),
                trade_tx: tx,
                active: false,
            })
        })
    }

    fn analyze_market_torch(&self, py: Python<'_>, data: Vec<f64>) -> PyResult<MarketData> {
        let torch = self.engine.torch.as_ref(py);
        let tensor = torch.call_method1("tensor", (data,))?;
        let cuda_tensor = tensor.call_method1("to", (&self.engine.device,))?;
        
        let price: f64 = cuda_tensor.call_method1("get", (0_i64,))?.call_method0("item")?.extract()?;
        let volume: f64 = cuda_tensor.call_method1("get", (1_i64,))?.call_method0("item")?.extract()?;
        
        Ok(MarketData {
            token: "".to_string(),
            price,
            volume_5m: volume / 288.0,
            volume_1h: volume / 24.0,
            liquidity: price * volume,
            buy_pressure: 0.0,
            volatility: 0.0,
            timestamp: chrono::Utc::now().timestamp(),
            stop_loss: price * self.engine.trade_config.emergency_stop_loss,
            take_profit: price * (1.0 + self.engine.trade_config.min_profit_target),
        })
    }

    fn monitor_positions<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {
        let engine = self.engine.clone();
        
        pyo3_asyncio::tokio::future_into_py(py, async move {
            let (tx, mut rx) = mpsc::channel::<Position>(100);
            let positions = engine.positions.clone();
            
            tokio::spawn(async move {
                while let Some(position) = rx.recv().await {
                    if let Ok(market_data) = engine.get_market_data(&position.token).await {
                        if market_data.price >= position.entry_price * (1.0 + engine.trade_config.min_profit_target) {
                            engine.close_position(&position, "take_profit").await.ok();
                        } else if market_data.price <= position.entry_price * engine.trade_config.emergency_stop_loss {
                            engine.close_position(&position, "stop_loss").await.ok();
                        }
                    }
                }
            });

            loop {
                let positions = positions.lock().await.clone();
                for position in positions {
                    tx.send(position).await.ok();
                }
                sleep(Duration::from_secs(1)).await;
            }
            
            #[allow(unreachable_code)]
            Ok(())
        })
    }

    fn start<'p>(&mut self, py: Python<'p>) -> PyResult<&'p PyAny> {
        self.active = true;
        let engine = self.engine.clone();
        let trade_tx = self.trade_tx.clone();

        // Start monitoring positions
        self.monitor_positions(py)?;

        pyo3_asyncio::tokio::future_into_py(py, async move {
            println!("🚀 TurboTrader Active - CUDA Enabled");

            while let Ok(market_data) = engine.process_pool_parallel().await {
                for data in market_data {
                    if data.liquidity >= MIN_LIQUIDITY {
                        if let Ok(signature) = engine.execute_trade(&data).await {
                            let sig_clone = signature.clone();
                            trade_tx.send(signature).unwrap_or_default();
                            println!("✅ Trade executed: {}", sig_clone);
                        }
                    }
                }
                sleep(Duration::from_millis(100)).await;
            }
            Ok(())
        })
    }

    fn stop(&mut self) -> PyResult<()> {
        self.active = false;
        println!("✨ TurboTrader Stopped");
        Ok(())
    }
}

#[pymodule]
fn solana_rust_bot(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_class::<TurboPyTrader>()?;
    Ok(())
} just uhm rewrite it with the code it needs to keep please