feat: implement NotionWriter database insertion and wire it with HyperliquidMonitor

src/hyperliquid.rs

@@ -202,6 +202,7 @@ impl HyperliquidMonitor {
                                         sz: format!("{:.5}", state.accumulated_sz),
                                         time: state.time,
                                         tid: state.tid,
+                                        oid,
                                         action: TradeAction::Open,
                                         start_pos: "0.0".to_string(),
                                         end_pos: format!("{:.5}", state.accumulated_sz),

src/lib.rs

@@ -1,5 +1,7 @@
 pub mod hyperliquid;
+pub mod notion;
 pub mod structs;
 
 pub use hyperliquid::{HyperliquidMonitor, HyperliquidMonitorError};
-pub use structs::{PositionTradeEvent, TradeAction};
+pub use notion::{NotionWriter, NotionWriterError};
+pub use structs::{NotionRowData, PositionTradeEvent, TradeAction};

src/main.rs

@@ -1,6 +1,7 @@
 use anyhow::{Context, Result};
+use chrono::{TimeZone, Utc};
 use dotenvy::dotenv;
-use nosync::HyperliquidMonitor;
+use nosync::{HyperliquidMonitor, NotionRowData, NotionWriter};
 use std::env;
 use tokio::sync::mpsc;
 use tracing::{error, info};
@@ -18,20 +19,29 @@ async fn main() -> Result<()> {
     tracing::subscriber::set_global_default(subscriber)
         .map_err(|e| anyhow::anyhow!("failed to set global tracing subscriber: {e}"))?;
 
-    info!("Starting Hyperliquid Monitor Service...");
+    info!("Starting Hyperliquid to Notion Sync Service...");
 
-    // Retrieve WALLET_ADDRESS from environment
+    // Retrieve WALLET_ADDRESS and IS_TESTNET 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}"))?;
 
-    // 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);
 
+    // Retrieve Notion configurations from environment
+    let notion_token =
+        env::var("NOTION_API_KEY").context("Missing NOTION_API_KEY in environment variables")?;
+    let notion_database_id = env::var("NOTION_DATABASE_ID")
+        .context("Missing NOTION_DATABASE_ID in environment variables")?;
+
+    // Initialize monitor and writer
     let monitor = HyperliquidMonitor::new(wallet, is_testnet);
+    let writer = NotionWriter::new(notion_token, notion_database_id)
+        .context("Failed to initialize NotionWriter")?;
+
     let (tx, mut rx) = mpsc::unbounded_channel();
 
     // Spawn the monitor loop in the background
@@ -41,21 +51,51 @@ async fn main() -> Result<()> {
         }
     });
 
-    info!("Service running, waiting for wallet events...");
+    info!("Service running. Monitoring position openings and writing to Notion...");
     while let Some(event) = rx.recv().await {
         info!(
             coin = %event.coin,
             side = %event.side,
             px = %event.px,
             sz = %event.sz,
-            action = ?event.action,
-            start_pos = %event.start_pos,
-            end_pos = %event.end_pos,
-            time = event.time,
-            tid = event.tid,
-            "Captured position trade event!"
+            oid = event.oid,
+            "Captured position open event, writing to Notion..."
         );
-        // Notion database integration will be wired here in the future
+
+        // Format Date/Time to "YYYY/MM/DD HH:MM"
+        let dt = Utc
+            .timestamp_millis_opt(event.time as i64)
+            .single()
+            .unwrap_or_else(Utc::now);
+        let date_time_str = dt.format("%Y/%m/%d %H:%M").to_string();
+
+        // Map Symbol to appends USDC
+        let symbol_formatted = format!("{}USDC", event.coin);
+
+        // Map Direction: B -> Long, S -> Short
+        let direction_str = if event.side == "B" {
+            "Long".to_string()
+        } else {
+            "Short".to_string()
+        };
+
+        // Construct Notion Row Data
+        let row = NotionRowData {
+            symbol: symbol_formatted,
+            quantity: event.sz,
+            filled_price: event.px,
+            direction: direction_str,
+            exchange: "Hyperliquid".to_string(), // Can be dynamically set or config-driven in the future
+            date_time: date_time_str,
+            time: event.time,
+            order_id: event.oid,
+            check: false,
+        };
+
+        // Write row to Notion
+        if let Err(e) = writer.write_row(row).await {
+            error!("Failed to write row to Notion: {:?}", e);
+        }
     }
 
     Ok(())

src/notion.rs

@@ -0,0 +1,179 @@
+use crate::structs::NotionRowData;
+use chrono::{TimeZone, Utc};
+use notion_client::endpoints::Client as NotionClient;
+use notion_client::endpoints::pages::create::request::CreateAPageRequestBuilder;
+use notion_client::objects::page::{DatePropertyValue, PageProperty, SelectPropertyValue};
+use notion_client::objects::parent::Parent;
+use notion_client::objects::property::DateOrDateTime;
+use notion_client::objects::rich_text::{RichText, Text};
+use reqwest::ClientBuilder;
+use serde_json::Number;
+use std::collections::BTreeMap;
+use thiserror::Error;
+use tracing::info;
+
+/// Errors specific to the NotionWriter.
+#[derive(Error, Debug)]
+pub enum NotionWriterError {
+    #[error("Notion client error: {0}")]
+    ClientError(String),
+    #[error("Notion API request failed: {0}")]
+    RequestError(#[from] notion_client::NotionClientError),
+}
+
+/// Writer for logging trade records to the Notion Database.
+pub struct NotionWriter {
+    client: NotionClient,
+    database_id: String,
+}
+
+impl NotionWriter {
+    /// Creates a new NotionWriter instance.
+    pub fn new(token: String, database_id: String) -> Result<Self, NotionWriterError> {
+        let client = NotionClient::new(token, Some(ClientBuilder::new()))
+            .map_err(|e| NotionWriterError::ClientError(format!("{:?}", e)))?;
+        Ok(Self {
+            client,
+            database_id,
+        })
+    }
+
+    /// Inserts a new row into the target Notion database.
+    pub async fn write_row(&self, data: NotionRowData) -> Result<(), NotionWriterError> {
+        info!(
+            symbol = %data.symbol,
+            order_id = data.order_id,
+            "Writing row to Notion database"
+        );
+
+        let mut properties = BTreeMap::new();
+
+        // 1. Symbol (Title column)
+        properties.insert(
+            "Symbol".to_string(),
+            PageProperty::Title {
+                id: None,
+                title: vec![RichText::Text {
+                    text: Text {
+                        content: data.symbol,
+                        link: None,
+                    },
+                    annotations: None,
+                    plain_text: None,
+                    href: None,
+                }],
+            },
+        );
+
+        // 2. Quantity (RichText column)
+        properties.insert(
+            "Quantity".to_string(),
+            PageProperty::RichText {
+                id: None,
+                rich_text: vec![RichText::Text {
+                    text: Text {
+                        content: data.quantity,
+                        link: None,
+                    },
+                    annotations: None,
+                    plain_text: None,
+                    href: None,
+                }],
+            },
+        );
+
+        // 3. Filled Price (RichText column)
+        properties.insert(
+            "Filled Price".to_string(),
+            PageProperty::RichText {
+                id: None,
+                rich_text: vec![RichText::Text {
+                    text: Text {
+                        content: data.filled_price,
+                        link: None,
+                    },
+                    annotations: None,
+                    plain_text: None,
+                    href: None,
+                }],
+            },
+        );
+
+        // 4. Direction (Select column)
+        properties.insert(
+            "Direction".to_string(),
+            PageProperty::Select {
+                id: None,
+                select: Some(SelectPropertyValue {
+                    id: None,
+                    name: Some(data.direction),
+                    color: None,
+                }),
+            },
+        );
+
+        // 5. Exchange (MultiSelect column)
+        properties.insert(
+            "Exchange".to_string(),
+            PageProperty::MultiSelect {
+                id: None,
+                multi_select: vec![SelectPropertyValue {
+                    id: None,
+                    name: Some(data.exchange),
+                    color: None,
+                }],
+            },
+        );
+
+        // 6. DataTime (Date column)
+        let dt = Utc
+            .timestamp_millis_opt(data.time as i64)
+            .single()
+            .unwrap_or_else(Utc::now);
+
+        properties.insert(
+            "DataTime".to_string(),
+            PageProperty::Date {
+                id: None,
+                date: Some(DatePropertyValue {
+                    start: Some(DateOrDateTime::DateTime(dt)),
+                    end: None,
+                    time_zone: None,
+                }),
+            },
+        );
+
+        // 7. Order ID (Number column)
+        properties.insert(
+            "Order ID".to_string(),
+            PageProperty::Number {
+                id: None,
+                number: Some(Number::from(data.order_id)),
+            },
+        );
+
+        // 8. Check (Checkbox column)
+        properties.insert(
+            "Check".to_string(),
+            PageProperty::Checkbox {
+                id: None,
+                checkbox: data.check,
+            },
+        );
+
+        // Build CreateAPageRequest and create page
+        let request = CreateAPageRequestBuilder::default()
+            .parent(Parent::DatabaseId {
+                database_id: self.database_id.clone(),
+            })
+            .properties(properties)
+            .build()
+            .map_err(|e| {
+                NotionWriterError::ClientError(format!("Failed to build page request: {:?}", e))
+            })?;
+
+        let page = self.client.pages.create_a_page(request).await?;
+        info!(page_id = %page.id, "Successfully wrote row to Notion Database");
+        Ok(())
+    }
+}

src/structs.rs

@@ -20,7 +20,22 @@ pub struct PositionTradeEvent {
     pub sz: String,   // Trade size as string
     pub time: u64,    // Epoch timestamp in milliseconds
     pub tid: u64,     // Unique trade ID
+    pub oid: u64,     // Hyperliquid Order ID
     pub action: TradeAction,
     pub start_pos: String, // Position size before the trade
     pub end_pos: String,   // Position size after the trade
 }
+
+/// Represents the structured row data formatted for Notion database insertion.
+#[derive(Debug, Clone)]
+pub struct NotionRowData {
+    pub symbol: String,       // e.g., "BTCUSDC"
+    pub quantity: String,     // e.g., "0.02484"
+    pub filled_price: String, // e.g., "72543"
+    pub direction: String,    // "Long" or "Short"
+    pub exchange: String,     // e.g., "Hyperliquid"
+    pub date_time: String,    // Format: "YYYY/MM/DD HH:MM"
+    pub time: u64,            // Unix timestamp in milliseconds
+    pub order_id: u64,        // Hyperliquid order ID (oid)
+    pub check: bool,          // false
+}