mirror of
https://github.com/exchanges-lab/tradesync.git
synced 2026-08-06 05:34:49 +08:00
feat: implement HyperliquidMonitor for position opening and migrate dependencies to git repository urls
This commit is contained in:
0xcathiefish
@@ -0,0 +1,144 @@
|
||||
use crate::structs::PositionOpenEvent;
|
||||
use alloy::primitives::Address;
|
||||
use hyperliquid_rust_sdk::{BaseUrl, InfoClient, Message, Subscription, UserData};
|
||||
use thiserror::Error;
|
||||
use tokio::sync::mpsc::UnboundedSender;
|
||||
use tokio::time::{Duration, sleep};
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
/// Errors specific to the HyperliquidMonitor.
|
||||
#[derive(Error, Debug)]
|
||||
pub enum HyperliquidMonitorError {
|
||||
#[error("Failed to parse wallet address: {0}")]
|
||||
AddressParseError(String),
|
||||
}
|
||||
|
||||
/// Monitor for Hyperliquid wallet position openings.
|
||||
pub struct HyperliquidMonitor {
|
||||
wallet_address: Address,
|
||||
is_testnet: bool,
|
||||
}
|
||||
|
||||
impl HyperliquidMonitor {
|
||||
/// Creates a new HyperliquidMonitor instance.
|
||||
pub fn new(wallet_address: Address, is_testnet: bool) -> Self {
|
||||
Self {
|
||||
wallet_address,
|
||||
is_testnet,
|
||||
}
|
||||
}
|
||||
|
||||
/// Runs the monitor WebSocket subscription in a loop, reconnecting if disconnected.
|
||||
pub async fn run(
|
||||
&self,
|
||||
event_tx: UnboundedSender<PositionOpenEvent>,
|
||||
) -> Result<(), HyperliquidMonitorError> {
|
||||
let base_url = if self.is_testnet {
|
||||
BaseUrl::Testnet
|
||||
} else {
|
||||
BaseUrl::Mainnet
|
||||
};
|
||||
|
||||
info!(
|
||||
wallet = %self.wallet_address,
|
||||
is_testnet = self.is_testnet,
|
||||
"Starting Hyperliquid monitor"
|
||||
);
|
||||
|
||||
loop {
|
||||
info!("Connecting to Hyperliquid InfoClient...");
|
||||
let mut info_client = match InfoClient::new(None, Some(base_url)).await {
|
||||
Ok(client) => client,
|
||||
Err(e) => {
|
||||
error!(
|
||||
error = ?e,
|
||||
"Failed to connect to Hyperliquid InfoClient, retrying in 5 seconds..."
|
||||
);
|
||||
sleep(Duration::from_secs(5)).await;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let (ws_sender, mut ws_receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
|
||||
info!(
|
||||
"Subscribing to UserEvents for wallet: {}",
|
||||
self.wallet_address
|
||||
);
|
||||
if let Err(e) = info_client
|
||||
.subscribe(
|
||||
Subscription::UserEvents {
|
||||
user: self.wallet_address,
|
||||
},
|
||||
ws_sender,
|
||||
)
|
||||
.await
|
||||
{
|
||||
error!(
|
||||
error = ?e,
|
||||
"Failed to subscribe to UserEvents, retrying in 5 seconds..."
|
||||
);
|
||||
sleep(Duration::from_secs(5)).await;
|
||||
continue;
|
||||
}
|
||||
|
||||
info!("Successfully subscribed to UserEvents.");
|
||||
|
||||
while let Some(msg) = ws_receiver.recv().await {
|
||||
match msg {
|
||||
Message::User(user_msg) => {
|
||||
debug!(user_msg = ?user_msg, "Received user event message");
|
||||
if let UserData::Fills(fills) = user_msg.data {
|
||||
for fill in fills {
|
||||
// Check if start_position is 0, which indicates a new position is opening
|
||||
let start_pos: f64 = fill.start_position.parse().unwrap_or(0.0);
|
||||
if start_pos == 0.0 {
|
||||
info!(
|
||||
coin = %fill.coin,
|
||||
side = %fill.side,
|
||||
px = %fill.px,
|
||||
sz = %fill.sz,
|
||||
tid = fill.tid,
|
||||
"Position opening detected!"
|
||||
);
|
||||
let event = PositionOpenEvent {
|
||||
coin: fill.coin,
|
||||
side: fill.side,
|
||||
px: fill.px,
|
||||
sz: fill.sz,
|
||||
time: fill.time,
|
||||
tid: fill.tid,
|
||||
};
|
||||
if let Err(e) = event_tx.send(event) {
|
||||
error!(
|
||||
error = ?e,
|
||||
"Failed to send PositionOpenEvent through channel"
|
||||
);
|
||||
}
|
||||
} else {
|
||||
debug!(
|
||||
coin = %fill.coin,
|
||||
start_pos = start_pos,
|
||||
"Ignore non-opening trade fill"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Message::HyperliquidError(err_msg) => {
|
||||
error!(error = %err_msg, "Received error message from Hyperliquid WS");
|
||||
}
|
||||
Message::Pong => {
|
||||
debug!("Received Pong from Hyperliquid WS");
|
||||
}
|
||||
other => {
|
||||
debug!(msg = ?other, "Received other message from Hyperliquid WS");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
warn!("Hyperliquid WebSocket connection closed. Reconnecting in 5 seconds...");
|
||||
sleep(Duration::from_secs(5)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
+3
-5
@@ -1,7 +1,5 @@
|
||||
pub mod module_a;
|
||||
pub mod module_b;
|
||||
pub mod hyperliquid;
|
||||
pub mod structs;
|
||||
|
||||
pub use module_a::{ModuleA, ModuleAError};
|
||||
pub use module_b::{ModuleB, ModuleBError};
|
||||
pub use structs::SharedMessage;
|
||||
pub use hyperliquid::{HyperliquidMonitor, HyperliquidMonitorError};
|
||||
pub use structs::PositionOpenEvent;
|
||||
|
||||
+37
-16
@@ -1,8 +1,9 @@
|
||||
use anyhow::{Context, Result};
|
||||
use dotenvy::dotenv;
|
||||
use nosync::{ModuleA, ModuleB};
|
||||
use nosync::HyperliquidMonitor;
|
||||
use std::env;
|
||||
use tracing::info;
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{error, info};
|
||||
use tracing_subscriber::{EnvFilter, FmtSubscriber};
|
||||
|
||||
#[tokio::main]
|
||||
@@ -10,29 +11,49 @@ async fn main() -> Result<()> {
|
||||
// Load environment variables from .env if present
|
||||
let _ = dotenv();
|
||||
|
||||
// Initialize tracing subscriber with settings from environment (RUST_LOG)
|
||||
// Initialize tracing subscriber
|
||||
let subscriber = FmtSubscriber::builder()
|
||||
.with_env_filter(EnvFilter::from_default_env())
|
||||
.finish();
|
||||
tracing::subscriber::set_global_default(subscriber)
|
||||
.map_err(|e| anyhow::anyhow!("failed to set global tracing subscriber: {e}"))?;
|
||||
|
||||
info!("Starting nosync application...");
|
||||
info!("Starting Hyperliquid Monitor Service...");
|
||||
|
||||
// Retrieve APP_NAME from environment variables, defaulting if not found
|
||||
let app_name = env::var("APP_NAME").unwrap_or_else(|_| "nosync-default".to_string());
|
||||
info!(app_name = %app_name, "Environment configured");
|
||||
// Retrieve WALLET_ADDRESS from environment
|
||||
let wallet_str =
|
||||
env::var("WALLET_ADDRESS").context("Missing WALLET_ADDRESS in environment variables")?;
|
||||
let wallet = wallet_str
|
||||
.parse::<alloy::primitives::Address>()
|
||||
.map_err(|e| anyhow::anyhow!("Invalid WALLET_ADDRESS: {e}"))?;
|
||||
|
||||
// Initialize ModuleA and ModuleB
|
||||
let module_a = ModuleA::new(app_name).context("Failed to initialize ModuleA")?;
|
||||
let module_b = ModuleB::new(module_a).context("Failed to initialize ModuleB")?;
|
||||
// Retrieve IS_TESTNET from environment, defaulting to true if not set
|
||||
let is_testnet_str = env::var("IS_TESTNET").unwrap_or_else(|_| "true".to_string());
|
||||
let is_testnet = is_testnet_str.parse::<bool>().unwrap_or(true);
|
||||
|
||||
// Run ModuleB logic
|
||||
module_b
|
||||
.run()
|
||||
.await
|
||||
.context("Error occurred during execution")?;
|
||||
let monitor = HyperliquidMonitor::new(wallet, is_testnet);
|
||||
let (tx, mut rx) = mpsc::unbounded_channel();
|
||||
|
||||
// Spawn the monitor loop in the background
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = monitor.run(tx).await {
|
||||
error!("Monitor runtime error: {:?}", e);
|
||||
}
|
||||
});
|
||||
|
||||
info!("Service running, waiting for wallet events...");
|
||||
while let Some(event) = rx.recv().await {
|
||||
info!(
|
||||
coin = %event.coin,
|
||||
side = %event.side,
|
||||
px = %event.px,
|
||||
sz = %event.sz,
|
||||
time = event.time,
|
||||
tid = event.tid,
|
||||
"Captured position opening event!"
|
||||
);
|
||||
// Notion database integration will be wired here in the future
|
||||
}
|
||||
|
||||
info!("nosync application finished successfully!");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
use crate::structs::SharedMessage;
|
||||
use thiserror::Error;
|
||||
use tracing::{debug, info};
|
||||
|
||||
/// Errors specific to ModuleA.
|
||||
#[derive(Error, Debug)]
|
||||
pub enum ModuleAError {
|
||||
#[error("Failed to process message: {0}")]
|
||||
ProcessError(String),
|
||||
}
|
||||
|
||||
/// ModuleA processes messages and manages a name field.
|
||||
pub struct ModuleA {
|
||||
name: String,
|
||||
}
|
||||
|
||||
impl ModuleA {
|
||||
/// Creates a new ModuleA instance.
|
||||
pub fn new(name: String) -> Result<Self, ModuleAError> {
|
||||
info!(name = %name, "Initializing ModuleA");
|
||||
Ok(Self { name })
|
||||
}
|
||||
|
||||
/// Processes a shared message asynchronously.
|
||||
pub async fn process_message(&self, msg: SharedMessage) -> Result<String, ModuleAError> {
|
||||
debug!(msg_id = msg.id, "Processing message in ModuleA");
|
||||
if msg.content.is_empty() {
|
||||
return Err(ModuleAError::ProcessError(
|
||||
"Empty message content".to_string(),
|
||||
));
|
||||
}
|
||||
Ok(format!("ModuleA[{}] processed: {}", self.name, msg.content))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_process_message_success() {
|
||||
let module_a = ModuleA::new("test-a".to_string()).unwrap();
|
||||
let msg = SharedMessage {
|
||||
id: 42,
|
||||
content: "Hello".to_string(),
|
||||
};
|
||||
let result = module_a.process_message(msg).await.unwrap();
|
||||
assert_eq!(result, "ModuleA[test-a] processed: Hello");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_process_message_empty_content() {
|
||||
let module_a = ModuleA::new("test-a".to_string()).unwrap();
|
||||
let msg = SharedMessage {
|
||||
id: 42,
|
||||
content: String::new(),
|
||||
};
|
||||
let result = module_a.process_message(msg).await;
|
||||
assert!(result.is_err());
|
||||
}
|
||||
}
|
||||
@@ -1,48 +0,0 @@
|
||||
use crate::module_a::ModuleA;
|
||||
use crate::structs::SharedMessage;
|
||||
use thiserror::Error;
|
||||
use tracing::info;
|
||||
|
||||
/// Errors specific to ModuleB.
|
||||
#[derive(Error, Debug)]
|
||||
pub enum ModuleBError {
|
||||
#[error("ModuleA error occurred: {0}")]
|
||||
AError(#[from] crate::module_a::ModuleAError),
|
||||
}
|
||||
|
||||
/// ModuleB holds a reference to ModuleA and interacts with it.
|
||||
pub struct ModuleB {
|
||||
processor: ModuleA,
|
||||
}
|
||||
|
||||
impl ModuleB {
|
||||
/// Creates a new ModuleB instance by taking ownership of a ModuleA processor.
|
||||
pub fn new(processor: ModuleA) -> Result<Self, ModuleBError> {
|
||||
info!("Initializing ModuleB");
|
||||
Ok(Self { processor })
|
||||
}
|
||||
|
||||
/// Runs the business logic of ModuleB.
|
||||
pub async fn run(&self) -> Result<(), ModuleBError> {
|
||||
let msg = SharedMessage {
|
||||
id: 100,
|
||||
content: "Hello from ModuleB".to_string(),
|
||||
};
|
||||
let result = self.processor.process_message(msg).await?;
|
||||
info!(result = %result, "ModuleB execution succeeded");
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_module_b_run() {
|
||||
let module_a = ModuleA::new("test-a".to_string()).unwrap();
|
||||
let module_b = ModuleB::new(module_a).unwrap();
|
||||
let result = module_b.run().await;
|
||||
assert!(result.is_ok());
|
||||
}
|
||||
}
|
||||
+8
-4
@@ -1,6 +1,10 @@
|
||||
/// Shared message structure across modules.
|
||||
/// Represents a detected position opening event on Hyperliquid.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SharedMessage {
|
||||
pub id: u64,
|
||||
pub content: String,
|
||||
pub struct PositionOpenEvent {
|
||||
pub coin: String,
|
||||
pub side: String, // "B" (Buy) or "S" (Sell)
|
||||
pub px: String, // Price as string
|
||||
pub sz: String, // Size as string
|
||||
pub time: u64, // Epoch timestamp in milliseconds
|
||||
pub tid: u64, // Unique trade/fill ID
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user