Periodically update unrealized pnl for position in database

coordinator/migrations/2023-07-19-055143_closed_positions/down.sql

@@ -4,4 +4,4 @@
 -- However, there is no proper way to replace the values to be removed where they are used (i.e. referenced in `positions` table)
 -- We opt to NOT remove enum values that were added at a later point.
 ALTER TABLE
-    "positions" DROP COLUMN "realized_pnl";
+    "positions" DROP COLUMN "realized_pnl_sat";

coordinator/migrations/2023-07-19-055143_closed_positions/up.sql

@@ -7,4 +7,4 @@ ADD
 ALTER TABLE
     positions
 ADD
-    COLUMN "realized_pnl" BIGINT;
+    COLUMN "realized_pnl_sat" BIGINT;

coordinator/migrations/2023-07-20-055142_unrealized_pnl_positions/down.sql

@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE
+    "positions" DROP COLUMN "unrealized_pnl_sat";

coordinator/migrations/2023-07-20-055142_unrealized_pnl_positions/up.sql

@@ -0,0 +1,5 @@
+-- Your SQL goes here
+ALTER TABLE
+    positions
+ADD
+    COLUMN "unrealized_pnl_sat" BIGINT;

coordinator/migrations/2023-07-21-050135_closing_price_positions/down.sql

@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE
+    "positions" DROP COLUMN "closing_price";

coordinator/migrations/2023-07-21-050135_closing_price_positions/up.sql

@@ -0,0 +1,5 @@
+-- Your SQL goes here
+ALTER TABLE
+    positions
+ADD
+    COLUMN "closing_price" REAL;

coordinator/src/bin/coordinator.rs

@@ -9,6 +9,7 @@ use coordinator::node::closed_positions;
 use coordinator::node::connection;
 use coordinator::node::expired_positions;
 use coordinator::node::storage::NodeStorage;
+use coordinator::node::unrealized_pnl;
 use coordinator::node::Node;
 use coordinator::routes::router;
 use coordinator::run_migration;
@@ -34,6 +35,7 @@ const PROCESS_PROMETHEUS_METRICS: Duration = Duration::from_secs(10);
 const PROCESS_INCOMING_DLC_MESSAGES_INTERVAL: Duration = Duration::from_secs(5);
 const EXPIRED_POSITION_SYNC_INTERVAL: Duration = Duration::from_secs(300);
 const CLOSED_POSITION_SYNC_INTERVAL: Duration = Duration::from_secs(30);
+const UNREALIZED_PNL_SYNC_INTERVAL: Duration = Duration::from_secs(600);
 const CONNECTION_CHECK_INTERVAL: Duration = Duration::from_secs(30);
 
 #[tokio::main]
@@ -162,6 +164,20 @@ async fn main() -> Result<()> {
         }
     });
 
+    tokio::spawn({
+        let node = node.clone();
+        async move {
+            loop {
+                tokio::time::sleep(UNREALIZED_PNL_SYNC_INTERVAL).await;
+                if let Err(e) = unrealized_pnl::sync(node.clone()).await {
+                    tracing::error!(
+                        "Failed to sync unrealized PnL with positions in database: {e:#}"
+                    );
+                }
+            }
+        }
+    });
+
     tokio::spawn({
         let node = node.clone();
         async move {

coordinator/src/db/positions.rs

@@ -32,7 +32,9 @@ pub struct Position {
     pub update_timestamp: OffsetDateTime,
     pub trader_pubkey: String,
     pub temporary_contract_id: Option<String>,
-    pub realized_pnl: Option<i64>,
+    pub realized_pnl_sat: Option<i64>,
+    pub unrealized_pnl_sat: Option<i64>,
+    pub closing_price: Option<f32>,
 }
 
 impl Position {
@@ -91,12 +93,14 @@ impl Position {
     pub fn set_open_position_to_closing(
         conn: &mut PgConnection,
         trader_pubkey: String,
+        closing_price: f32,
     ) -> Result<()> {
         let effected_rows = diesel::update(positions::table)
             .filter(positions::trader_pubkey.eq(trader_pubkey.clone()))
             .filter(positions::position_state.eq(PositionState::Open))
             .set((
                 positions::position_state.eq(PositionState::Closing),
+                positions::closing_price.eq(Some(closing_price)),
                 positions::update_timestamp.eq(OffsetDateTime::now_utc()),
             ))
             .execute(conn)?;
@@ -113,7 +117,7 @@ impl Position {
             .filter(positions::id.eq(id))
             .set((
                 positions::position_state.eq(PositionState::Closed),
-                positions::realized_pnl.eq(Some(pnl)),
+                positions::realized_pnl_sat.eq(Some(pnl)),
                 positions::update_timestamp.eq(OffsetDateTime::now_utc()),
             ))
             .execute(conn)?;
@@ -125,6 +129,22 @@ impl Position {
         Ok(())
     }
 
+    pub fn update_unrealized_pnl(conn: &mut PgConnection, id: i32, pnl: i64) -> Result<()> {
+        let effected_rows = diesel::update(positions::table)
+            .filter(positions::id.eq(id))
+            .set((
+                positions::unrealized_pnl_sat.eq(Some(pnl)),
+                positions::update_timestamp.eq(OffsetDateTime::now_utc()),
+            ))
+            .execute(conn)?;
+
+        if effected_rows == 0 {
+            bail!("Could not update unrealized pnl {pnl} for position {id}")
+        }
+
+        Ok(())
+    }
+
     /// inserts the given position into the db. Returns the position if successful
     #[autometrics]
     pub fn insert(
@@ -151,7 +171,7 @@ impl From<Position> for crate::position::models::Position {
             liquidation_price: value.liquidation_price,
             position_state: crate::position::models::PositionState::from((
                 value.position_state,
-                value.realized_pnl,
+                value.realized_pnl_sat,
             )),
             collateral: value.collateral,
             creation_timestamp: value.creation_timestamp,
@@ -161,6 +181,7 @@ impl From<Position> for crate::position::models::Position {
             temporary_contract_id: value.temporary_contract_id.map(|contract_id| {
                 ContractId::from_hex(contract_id.as_str()).expect("contract id to decode")
             }),
+            closing_price: value.closing_price,
         }
     }
 }

coordinator/src/node.rs

@@ -54,9 +54,10 @@ pub mod expired_positions;
 pub mod order_matching_fee;
 pub mod routing_fees;
 pub mod storage;
+pub mod unrealized_pnl;
 
 /// The leverage used by the coordinator for all trades.
-const COORDINATOR_LEVERAGE: f32 = 1.0;
+pub const COORDINATOR_LEVERAGE: f32 = 1.0;
 
 #[derive(Debug, Clone)]
 pub struct NodeSettings {
@@ -332,6 +333,9 @@ impl Node {
         db::positions::Position::set_open_position_to_closing(
             &mut connection,
             position.trader.to_string(),
+            closing_price
+                .to_f32()
+                .expect("Closing price to fit into f32"),
         )
     }
 

coordinator/src/node/expired_positions.rs

@@ -57,10 +57,7 @@ pub async fn close(node: Node) {
         };
 
         let closing_price = match BitmexClient::get_quote(&position.expiry_timestamp).await {
-            Ok(quote) => match position.direction {
-                trade::Direction::Long => quote.bid_price,
-                trade::Direction::Short => quote.ask_price,
-            },
+            Ok(quote) => quote.get_price_for_direction(position.direction.opposite()),
             Err(e) => {
                 tracing::warn!(
                     "Failed to get quote from bitmex for {} at {}. Error: {e:?}",

coordinator/src/node/unrealized_pnl.rs

@@ -0,0 +1,40 @@
+use crate::db;
+use crate::node::Node;
+use crate::position::models::Position;
+use anyhow::Context;
+use anyhow::Result;
+use diesel::r2d2::ConnectionManager;
+use diesel::r2d2::PooledConnection;
+use diesel::PgConnection;
+use time::OffsetDateTime;
+use trade::bitmex_client::BitmexClient;
+use trade::bitmex_client::Quote;
+
+pub async fn sync(node: Node) -> Result<()> {
+    let mut conn = node.pool.get()?;
+
+    let positions = db::positions::Position::get_all_open_or_closing_positions(&mut conn)?;
+    let current_quote = BitmexClient::get_quote(&OffsetDateTime::now_utc())
+        .await
+        .context("Failed to fetch quote from BitMEX")?;
+
+    for position in positions.iter() {
+        if let Err(e) = sync_position(&mut conn, position, current_quote.clone()) {
+            tracing::error!(position_id=%position.id, ?current_quote, "Failed to update position's unrealized pnl in database: {e:#}")
+        }
+    }
+
+    Ok(())
+}
+
+fn sync_position(
+    conn: &mut PooledConnection<ConnectionManager<PgConnection>>,
+    position: &Position,
+    quote: Quote,
+) -> Result<()> {
+    let pnl = position.calculate_coordinator_pnl(quote)?;
+    db::positions::Position::update_unrealized_pnl(conn, position.id, pnl)
+        .context("Failed to update unrealized pnl in db")?;
+
+    Ok(())
+}