diff --git a/.env.example b/.env.example index 5d27bb9..8485a43 100644 --- a/.env.example +++ b/.env.example @@ -40,3 +40,8 @@ BACKTEST_CANDLES=50000 # ── Live Trading ───────────────────────────────────────────────────────────── # LIVE=true # LIVE_POLL_SECS=30 + +# ── Telegram Notifications ─────────────────────────────────────────────────── +# TELEGRAM_BOT_TOKEN=123456:ABC-xxxx +# TELEGRAM_CHAT_ID=-100xxxxxxxxxx +# TELEGRAM_THREAD_ID=123 # optional, for forum topics/threads diff --git a/Cargo.lock b/Cargo.lock index f6e4078..84e6d94 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -45,7 +45,9 @@ dependencies = [ "chrono", "domain", "dotenvy", + "futures-util", "mt5-client", + "reqwest", "rust_decimal", "serde", "serde_json", @@ -286,6 +288,29 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + [[package]] name = "futures-task" version = "0.3.32" @@ -299,7 +324,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -994,6 +1023,7 @@ dependencies = [ "base64", "bytes", "futures-core", + "futures-util", "http", "http-body", "http-body-util", @@ -1013,12 +1043,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", "webpki-roots", ] @@ -1414,6 +1446,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -1688,6 +1733,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "web-sys" version = "0.3.100" diff --git a/Cargo.toml b/Cargo.toml index 4774aa5..51a2a35 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,8 @@ path = "src/main.rs" anyhow = "1" chrono = { version = "0.4", features = ["serde"] } dotenvy = "0.15" +futures-util = "0.3" +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "stream"] } rust_decimal = { version = "1", features = ["serde"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/crates/domain/src/deal.rs b/crates/domain/src/deal.rs new file mode 100644 index 0000000..17f0dd6 --- /dev/null +++ b/crates/domain/src/deal.rs @@ -0,0 +1,26 @@ +use chrono::{DateTime, Utc}; +use rust_decimal::Decimal; +use serde::Deserialize; + +use crate::serde_helpers; + +#[derive(Debug, Clone, Deserialize)] +pub struct Deal { + pub ticket: u64, + #[serde(with = "serde_helpers::naive_utc_secs")] + pub time: DateTime, + pub entry: u8, // 0 = open, 1 = close + pub magic: u64, + #[serde(deserialize_with = "serde_helpers::de_decimal")] + pub volume: Decimal, + #[serde(deserialize_with = "serde_helpers::de_decimal")] + pub price: Decimal, + #[serde(deserialize_with = "serde_helpers::de_decimal")] + pub profit: Decimal, + #[serde(deserialize_with = "serde_helpers::de_decimal")] + pub commission: Decimal, + #[serde(deserialize_with = "serde_helpers::de_decimal")] + pub swap: Decimal, + pub symbol: String, + pub comment: String, +} diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 7a42adc..6ac3e62 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -2,6 +2,7 @@ mod serde_helpers; pub mod account; pub mod candle; +pub mod deal; pub mod position; pub mod symbol; pub mod tick; @@ -9,6 +10,7 @@ pub mod timeframe; pub use account::AccountInfo; pub use candle::Candle; +pub use deal::Deal; pub use position::{Position, Side}; pub use symbol::Symbol; pub use tick::Tick; diff --git a/crates/mt5-client/src/client.rs b/crates/mt5-client/src/client.rs index 0afa165..0ce5af1 100644 --- a/crates/mt5-client/src/client.rs +++ b/crates/mt5-client/src/client.rs @@ -122,6 +122,23 @@ impl Mt5Client { Ok(w.data) } + pub async fn history_deals( + &self, + date_from: &str, + date_to: &str, + symbol: Option<&str>, + ) -> Result, Mt5Error> { + let url = format!("{}/history/deals", self.base_url); + let mut req = self.http.get(&url) + .query(&[("date_from", date_from), ("date_to", date_to)]); + if let Some(sym) = symbol { + req = req.query(&[("symbol", sym)]); + } + let text = self.fetch_text(req).await?; + let w: DataVec = serde_json::from_str(&text)?; + Ok(w.data) + } + pub async fn place_order( &self, request: &TradeRequest, diff --git a/crates/mt5-client/src/lib.rs b/crates/mt5-client/src/lib.rs index b6a58b4..e5bea09 100644 --- a/crates/mt5-client/src/lib.rs +++ b/crates/mt5-client/src/lib.rs @@ -4,4 +4,5 @@ mod types; pub use client::Mt5Client; pub use error::Mt5Error; +pub use domain::Deal; pub use types::{HealthStatus, OrderCheckResult, PendingOrder, TradeRequest, TradeResult}; diff --git a/src/live.rs b/src/live.rs index 2e76696..d830903 100644 --- a/src/live.rs +++ b/src/live.rs @@ -1,55 +1,62 @@ use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use domain::{Side, Timeframe}; +use futures_util::StreamExt; use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; -use std::path::PathBuf; +use std::{path::PathBuf, sync::Arc}; use tokio::time::{interval, Duration}; -use crate::{detector, helpers::{d2f, fmt_price, rolling_ema, size_position}}; +use crate::{ + detector, + helpers::{d2f, fmt_price, rolling_ema, size_position}, + telegram::TelegramConfig, +}; -// Magic number that identifies all Ares orders/positions in MT5. -const MAGIC: u64 = 19730; - -// How many recent candles to fetch per tick (must cover EMA warm-up + 3 for FVG). +const MAGIC: u64 = 19730; const CANDLE_FETCH: u32 = 100; // ── config ──────────────────────────────────────────────────────────────────── #[derive(Debug, Clone)] pub struct LiveConfig { - pub symbol: String, - pub timeframe: Timeframe, - pub risk_pct: Decimal, - pub body_pct_min: Decimal, - pub close_pct_min: Decimal, + pub symbol: String, + pub timeframe: Timeframe, + pub risk_pct: Decimal, + pub body_pct_min: Decimal, + pub close_pct_min: Decimal, pub fvg_expiry_candles: usize, - pub min_fvg_pips: Decimal, - pub min_sl_pips: Decimal, - pub sl_buffer: Decimal, - pub min_rr: Decimal, - pub slippage_points: Decimal, - pub spread_override: Option, - pub ema_period: usize, - pub poll_secs: u64, + pub min_fvg_pips: Decimal, + pub min_sl_pips: Decimal, + pub sl_buffer: Decimal, + pub min_rr: Decimal, + pub slippage_points: Decimal, + pub spread_override: Option, + pub ema_period: usize, + pub poll_secs: u64, + pub mt5_base_url: String, + pub telegram: Option, } -// ── persisted state ─────────────────────────────────────────────────────────── +// ── pending order state ─────────────────────────────────────────────────────── #[derive(Debug, Serialize, Deserialize)] struct State { - ticket: u64, - expires_at: DateTime, + ticket: u64, + expires_at: DateTime, + tg_message_id: Option, + side: String, + entry: f64, + sl: f64, + tp: f64, + volume: f64, } impl State { - fn path(symbol: &str) -> PathBuf { - PathBuf::from(format!(".ares_state_{symbol}.json")) - } + fn path(symbol: &str) -> PathBuf { PathBuf::from(format!(".ares_state_{symbol}.json")) } fn load(symbol: &str) -> Option { - let content = std::fs::read_to_string(Self::path(symbol)).ok()?; - serde_json::from_str(&content).ok() + serde_json::from_str(&std::fs::read_to_string(Self::path(symbol)).ok()?).ok() } fn save(&self, symbol: &str) -> Result<()> { @@ -57,9 +64,52 @@ impl State { Ok(()) } - fn clear(symbol: &str) { - let _ = std::fs::remove_file(Self::path(symbol)); + fn clear(symbol: &str) { let _ = std::fs::remove_file(Self::path(symbol)); } +} + +// ── open position state (persisted for SSE notifications after fill) ────────── + +#[derive(Debug, Serialize, Deserialize)] +struct PosState { + ticket: u64, + tg_message_id: Option, + side: String, + entry: f64, + sl: f64, + tp: f64, + volume: f64, +} + +impl PosState { + fn path(symbol: &str) -> PathBuf { PathBuf::from(format!(".ares_pos_{symbol}.json")) } + + fn load(symbol: &str) -> Option { + serde_json::from_str(&std::fs::read_to_string(Self::path(symbol)).ok()?).ok() } + + fn save(&self, symbol: &str) -> Result<()> { + std::fs::write(Self::path(symbol), serde_json::to_string_pretty(self)?)?; + Ok(()) + } + + fn clear(symbol: &str) { let _ = std::fs::remove_file(Self::path(symbol)); } +} + +// ── SSE position event ──────────────────────────────────────────────────────── + +#[derive(Debug, Deserialize)] +struct SsePosition { + ticket: u64, + symbol: String, + #[serde(rename = "type")] + pos_type: u32, + volume: f64, + price_open: f64, + price_current: Option, + sl: f64, + tp: f64, + profit: f64, + magic: u64, } // ── entry point ─────────────────────────────────────────────────────────────── @@ -79,20 +129,30 @@ pub async fn run(mt5: &mt5_client::Mt5Client, cfg: &LiveConfig) -> Result<()> { let spread = cfg.spread_override .unwrap_or_else(|| Decimal::from(sym_info.spread) * point); - let tf_mins = timeframe_minutes(cfg.timeframe); + let tf_mins = timeframe_minutes(cfg.timeframe); let expiry_dur = chrono::Duration::minutes(tf_mins * cfg.fvg_expiry_candles as i64); + // spawn SSE position listener + if cfg.telegram.is_some() { + let mt5_arc = Arc::new(mt5_client::Mt5Client::new(cfg.mt5_base_url.clone())); + let http_arc = Arc::new(reqwest::Client::new()); + let symbol = cfg.symbol.clone(); + let tg = cfg.telegram.clone(); + let base_url = cfg.mt5_base_url.clone(); + tokio::spawn(async move { + sse_task(mt5_arc, http_arc, base_url, symbol, tg).await; + }); + } + + let http = reqwest::Client::new(); let mut ticker = interval(Duration::from_secs(cfg.poll_secs)); loop { ticker.tick().await; - if let Err(e) = tick( - mt5, cfg, &sym_info, contract_size, point, prec, pip_size, min_sl, min_zone, - slip, spread, profit_is_usd, expiry_dur, - ) - .await - { + mt5, cfg, &http, &sym_info, contract_size, point, prec, + pip_size, min_sl, min_zone, slip, spread, profit_is_usd, expiry_dur, + ).await { tracing::error!("tick error: {e:#}"); } } @@ -100,9 +160,11 @@ pub async fn run(mt5: &mt5_client::Mt5Client, cfg: &LiveConfig) -> Result<()> { // ── single poll tick ────────────────────────────────────────────────────────── +#[allow(clippy::too_many_arguments)] async fn tick( mt5: &mt5_client::Mt5Client, cfg: &LiveConfig, + http: &reqwest::Client, sym_info: &domain::Symbol, contract_size: Decimal, _point: Decimal, @@ -117,20 +179,34 @@ async fn tick( ) -> Result<()> { let symbol = &cfg.symbol; - // ── 1. check for open positions by this bot ─────────────────────────────── + // ── 1. open position? ───────────────────────────────────────────────────── let positions = mt5.positions().await.context("fetch positions")?; - let has_position = positions - .iter() - .any(|p| p.symbol == *symbol && p.magic == MAGIC); + let has_position = positions.iter().any(|p| p.symbol == *symbol && p.magic == MAGIC); if has_position { - tracing::debug!(%symbol, "position already open — skip"); + // ensure PosState exists so SSE task can find it on close + if PosState::load(symbol).is_none() { + if let Some(pos) = positions.iter().find(|p| p.symbol == *symbol && p.magic == MAGIC) { + let tg_msg_id = State::load(symbol).and_then(|s| s.tg_message_id); + let ps = PosState { + ticket: pos.ticket, + tg_message_id: tg_msg_id, + side: format!("{:?}", pos.side), + entry: d2f(pos.price_open), + sl: d2f(pos.sl), + tp: d2f(pos.tp), + volume: d2f(pos.volume), + }; + let _ = ps.save(symbol); + } + } + tracing::debug!(%symbol, "position open — skip"); return Ok(()); } - // ── 2. manage pending order state ──────────────────────────────────────── + // ── 2. manage pending order ─────────────────────────────────────────────── if let Some(state) = State::load(symbol) { - let orders = mt5.orders(symbol).await.context("fetch orders")?; + let orders = mt5.orders(symbol).await.context("fetch orders")?; let still_pending = orders.iter().any(|o| o.ticket == state.ticket && o.magic == MAGIC); if still_pending { @@ -139,65 +215,58 @@ async fn tick( return Ok(()); } // expired — cancel - tracing::info!(%symbol, ticket = state.ticket, "FVG setup expired — cancelling order"); + tracing::info!(%symbol, ticket = state.ticket, "FVG setup expired — cancelling"); match mt5.cancel_order(state.ticket, symbol).await { - Ok(r) => tracing::info!(retcode = r.retcode, "cancel ok"), + Ok(r) => tracing::info!(retcode = r.retcode, "cancel ok"), Err(e) => tracing::warn!("cancel failed: {e:#}"), } + if let (Some(tg), Some(msg_id)) = (&cfg.telegram, state.tg_message_id) { + let text = format!( + "⏱ EXPIRED\n{} {}\nEntry: {:.5}\nNo fill after {} candles", + symbol, state.side, state.entry, + cfg.fvg_expiry_candles, + ); + let _ = tg.edit(http, msg_id, &text).await; + } } else { - tracing::info!(%symbol, ticket = state.ticket, "pending order no longer in MT5 (filled/cancelled externally)"); + tracing::info!(%symbol, ticket = state.ticket, "order gone from MT5 (filled or cancelled)"); + // SSE task handles "Filled" notification when position opens } State::clear(symbol); return Ok(()); } - // ── 3. fetch recent candles ─────────────────────────────────────────────── + // ── 3. fetch candles ────────────────────────────────────────────────────── let candles = mt5 .rates_from_pos(symbol, cfg.timeframe, 0, CANDLE_FETCH) .await .context("fetch candles")?; - if candles.len() < 5 { - tracing::warn!(%symbol, "too few candles"); - return Ok(()); - } - + if candles.len() < 5 { return Ok(()); } let n = candles.len(); - // Use last 3 fully-closed bars: [n-4], [n-3], [n-2] — skip [n-1] which may - // still be forming at poll time. - let pre = &candles[n - 4]; - let impulse = &candles[n - 3]; - let post = &candles[n - 2]; - let last_idx = n - 4; // detector uses absolute index only for expiry, we don't need it + let pre = &candles[n - 4]; + let impulse = &candles[n - 3]; + let post = &candles[n - 2]; - // ── 4. EMA trend filter ─────────────────────────────────────────────────── + // ── 4. EMA filter ───────────────────────────────────────────────────────── let ema_val: Option = if cfg.ema_period > 0 && candles.len() >= cfg.ema_period { let closes: Vec = candles.iter().map(|c| c.close).collect(); - let emas = rolling_ema(&closes, cfg.ema_period); - emas[n - 2] + rolling_ema(&closes, cfg.ema_period)[n - 2] } else { Some(Decimal::ZERO) }; - // ── 5. detect momentum FVG ──────────────────────────────────────────────── - let fvg = detector::detect( - pre, - impulse, - post, - cfg.body_pct_min, - cfg.close_pct_min, - min_zone, - last_idx, - cfg.fvg_expiry_candles, - ); - - let fvg = match fvg { + // ── 5. detect FVG ───────────────────────────────────────────────────────── + let fvg = match detector::detect( + pre, impulse, post, + cfg.body_pct_min, cfg.close_pct_min, min_zone, + n - 4, cfg.fvg_expiry_candles, + ) { Some(f) => f, None => return Ok(()), }; - // EMA filter let ema_ok = match ema_val { Some(ema) => match fvg.side { Side::Long => post.close > ema, @@ -205,63 +274,46 @@ async fn tick( }, None => false, }; - if !ema_ok { - tracing::debug!(%symbol, ?fvg.side, "EMA filter rejected FVG"); - return Ok(()); - } + if !ema_ok { return Ok(()); } - // ── 6. compute SL / TP ─────────────────────────────────────────────────── + // ── 6. SL / TP ──────────────────────────────────────────────────────────── let sl = match fvg.side { Side::Long => fvg.impulse_sl - cfg.sl_buffer, Side::Short => fvg.impulse_sl + cfg.sl_buffer, }; let sl_dist = (fvg.entry - sl).abs(); - if sl_dist < min_sl { - tracing::debug!(%symbol, %sl_dist, "SL too tight — skip"); - return Ok(()); - } + if sl_dist < min_sl { return Ok(()); } let tp = match fvg.side { Side::Long => fvg.entry + sl_dist * cfg.min_rr, Side::Short => fvg.entry - sl_dist * cfg.min_rr, }; - // ── 7. size position ────────────────────────────────────────────────────── - let acct = mt5.account().await.context("fetch account")?; - let balance = Decimal::try_from(acct.balance).context("balance conversion")?; - - let ref_price = post.close; - let value_per_lot = if profit_is_usd || ref_price == Decimal::ZERO { + // ── 7. position size ────────────────────────────────────────────────────── + let acct = mt5.account().await.context("fetch account")?; + let balance = Decimal::try_from(acct.balance).context("balance")?; + let value_per_lot = if profit_is_usd || post.close == Decimal::ZERO { contract_size } else { - contract_size / ref_price + contract_size / post.close }; - let volume = match size_position( balance, cfg.risk_pct, sl_dist, value_per_lot, sym_info.volume_step, sym_info.volume_min, sym_info.volume_max, ) { Some(v) => v, - None => { - tracing::warn!(%symbol, "position sizing returned None (SL=0 or too small)"); - return Ok(()); - } + None => return Ok(()), }; - // ── 8. place pending limit order ────────────────────────────────────────── - let entry_price = fvg.entry; + // ── 8. place order ──────────────────────────────────────────────────────── let req = mt5_client::TradeRequest::limit( - fvg.side, symbol.clone(), d2f(volume), d2f(entry_price), d2f(sl), d2f(tp), + fvg.side, symbol.clone(), d2f(volume), d2f(fvg.entry), d2f(sl), d2f(tp), MAGIC, format!("ares-{}", post.time.format("%m%d-%H%M")), ); tracing::info!( %symbol, side = ?fvg.side, - entry = %fmt_price(entry_price, prec), - sl = %fmt_price(sl, prec), - tp = %fmt_price(tp, prec), - vol = %volume, - bal = %balance, - "placing limit order", + entry = %fmt_price(fvg.entry, prec), sl = %fmt_price(sl, prec), tp = %fmt_price(tp, prec), + vol = %volume, bal = %balance, "placing limit order", ); let result = mt5.place_order(&req).await.context("place_order")?; @@ -269,15 +321,234 @@ async fn tick( tracing::error!(retcode = result.retcode, comment = %result.comment, "order rejected"); return Ok(()); } - tracing::info!(ticket = result.order, "order placed"); + let side_str = format!("{:?}", fvg.side); + let rr = d2f(cfg.min_rr); + + // send Telegram "Pending" message + let tg_msg_id = if let Some(tg) = &cfg.telegram { + let text = format!( + "🟡 PENDING\n{} {}\nEntry: {}\nSL: {} TP: {}\nVol: {} lot RR: {:.1}", + symbol, side_str, + fmt_price(fvg.entry, prec), fmt_price(sl, prec), fmt_price(tp, prec), + volume, rr, + ); + match tg.send(http, &text).await { + Ok(id) => { tracing::info!(msg_id = id, "Telegram pending sent"); Some(id) } + Err(e) => { tracing::warn!("Telegram send failed: {e:#}"); None } + } + } else { None }; + let state = State { - ticket: result.order, - expires_at: Utc::now() + expiry_dur, + ticket: result.order, + expires_at: Utc::now() + expiry_dur, + tg_message_id: tg_msg_id, + side: side_str, + entry: d2f(fvg.entry), + sl: d2f(sl), + tp: d2f(tp), + volume: d2f(volume), }; state.save(symbol).context("save state")?; + Ok(()) +} +// ── SSE position listener ───────────────────────────────────────────────────── + +async fn sse_task( + mt5: Arc, + http: Arc, + base_url: String, + symbol: String, + tg: Option, +) { + loop { + tracing::debug!(%symbol, "SSE connecting"); + if let Err(e) = sse_loop(&mt5, &http, &base_url, &symbol, &tg).await { + tracing::warn!(%symbol, "SSE error: {e:#}"); + } + tokio::time::sleep(Duration::from_secs(5)).await; + } +} + +async fn sse_loop( + mt5: &mt5_client::Mt5Client, + http: &reqwest::Client, + base_url: &str, + symbol: &str, + tg: &Option, +) -> Result<()> { + let url = format!("{}/positions/stream?symbol={}", base_url, symbol); + let resp = http.get(&url).send().await?; + if !resp.status().is_success() { + anyhow::bail!("SSE stream status {}", resp.status()); + } + + let mut stream = resp.bytes_stream(); + let mut buf = String::new(); + let mut event_type = String::new(); + + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + buf.push_str(&String::from_utf8_lossy(&chunk)); + + loop { + match buf.find('\n') { + None => break, + Some(pos) => { + let line = buf[..pos].trim_end_matches('\r').to_string(); + buf.drain(..=pos); + + if line.starts_with("event:") { + event_type = line[6..].trim().to_string(); + } else if line.starts_with("data:") && !event_type.is_empty() { + let data = line[5..].trim().to_string(); + handle_sse_event(mt5, http, symbol, &event_type, &data, tg).await; + event_type.clear(); + } else if line.is_empty() { + event_type.clear(); + } + } + } + } + } + Ok(()) +} + +async fn handle_sse_event( + mt5: &mt5_client::Mt5Client, + http: &reqwest::Client, + symbol: &str, + event_type: &str, + data: &str, + tg: &Option, +) { + let pos: SsePosition = match serde_json::from_str(data) { + Ok(p) => p, + Err(e) => { tracing::warn!("SSE parse error: {e:#}"); return; } + }; + + if pos.magic != MAGIC || pos.symbol != symbol { return; } + + match event_type { + "position_opened" => on_position_opened(http, symbol, &pos, tg).await, + "position_closed" => on_position_closed(mt5, http, symbol, &pos, tg).await, + _ => {} + } +} + +async fn on_position_opened( + http: &reqwest::Client, + symbol: &str, + pos: &SsePosition, + tg: &Option, +) { + tracing::info!(%symbol, ticket = pos.ticket, "position opened (SSE)"); + + // promote pending state → position state + let state = State::load(symbol); + let tg_msg_id = state.as_ref().and_then(|s| s.tg_message_id); + + let ps = PosState { + ticket: pos.ticket, + tg_message_id: tg_msg_id, + side: if pos.pos_type == 0 { "Long".to_string() } else { "Short".to_string() }, + entry: pos.price_open, + sl: pos.sl, + tp: pos.tp, + volume: pos.volume, + }; + let _ = ps.save(symbol); + State::clear(symbol); + + if let (Some(tg), Some(msg_id)) = (tg, tg_msg_id) { + let side_str = &ps.side; + let text = format!( + "⚡ FILLED\n{} {}\nEntry: {:.5}\nSL: {:.5} TP: {:.5}\nVol: {:.2} lot", + symbol, side_str, pos.price_open, pos.sl, pos.tp, pos.volume, + ); + let _ = tg.edit(http, msg_id, &text).await; + } +} + +async fn on_position_closed( + mt5: &mt5_client::Mt5Client, + http: &reqwest::Client, + symbol: &str, + pos: &SsePosition, + tg: &Option, +) { + tracing::info!(%symbol, ticket = pos.ticket, profit = pos.profit, "position closed (SSE)"); + + let ps = PosState::load(symbol); + PosState::clear(symbol); + + let Some(tg) = tg else { return }; + let Some(msg_id) = ps.as_ref().and_then(|s| s.tg_message_id) else { return }; + + let (icon, label) = if pos.profit >= 0.0 { ("✅", "TP HIT") } else { ("❌", "SL HIT") }; + let exit = pos.price_current.unwrap_or(0.0); + let entry = ps.as_ref().map(|s| s.entry).unwrap_or(pos.price_open); + let side_str = ps.as_ref().map(|s| s.side.as_str()).unwrap_or("?"); + + let acct_bal = mt5.account().await.ok().map(|a| a.balance).unwrap_or(Decimal::ZERO); + + let text = format!( + "{} {} {:+.2}\n{} {}\n{:.5} → {:.5}\nVol: {:.2} lot\nBal: ${}", + icon, label, pos.profit, + symbol, side_str, + entry, exit, + pos.volume, acct_bal, + ); + let _ = tg.edit(http, msg_id, &text).await; + + // send PnL summary after close + let _ = send_pnl_summary(mt5, http, symbol, tg).await; +} + +// ── PnL summary ─────────────────────────────────────────────────────────────── + +async fn send_pnl_summary( + mt5: &mt5_client::Mt5Client, + http: &reqwest::Client, + symbol: &str, + tg: &TelegramConfig, +) -> Result<()> { + let now = Utc::now(); + let today_str = now.format("%Y-%m-%dT00:00:00").to_string(); + let now_str = now.format("%Y-%m-%dT%H:%M:%S").to_string(); + let epoch_str = "2020-01-01T00:00:00".to_string(); + + let (today_deals, all_deals) = tokio::try_join!( + mt5.history_deals(&today_str, &now_str, Some(symbol)), + mt5.history_deals(&epoch_str, &now_str, Some(symbol)), + )?; + + let summary_text = |label: &str, deals: &[domain::Deal]| { + let closing: Vec<_> = deals.iter() + .filter(|d| d.entry == 1 && d.magic == MAGIC) + .collect(); + let total = closing.len(); + let wins = closing.iter().filter(|d| d.profit > Decimal::ZERO).count(); + let losses = total - wins; + let profit: Decimal = closing.iter().map(|d| d.profit + d.commission + d.swap).sum(); + let wr = if total > 0 { wins as f64 / total as f64 * 100.0 } else { 0.0 }; + format!( + "{}\nTrades: {} ({}W / {}L) WR: {:.0}%\nProfit: ${:+.2}", + label, total, wins, losses, wr, profit, + ) + }; + + let today_date = now.format("%Y-%m-%d").to_string(); + let text = format!( + "📊 Today — {} {}\n{}\n\n📈 All-time\n{}", + symbol, today_date, + summary_text("", &today_deals), + summary_text("", &all_deals), + ); + + tg.send(http, &text).await?; Ok(()) } @@ -308,4 +579,3 @@ fn timeframe_minutes(tf: Timeframe) -> i64 { Timeframe::Mn1 => 43200, } } - diff --git a/src/main.rs b/src/main.rs index dabd0e5..8c5eb85 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ mod backtest; mod detector; mod helpers; mod live; +mod telegram; use anyhow::Context; use chrono::NaiveDate; @@ -90,12 +91,26 @@ async fn main() -> anyhow::Result<()> { let spread_override = env_spread_override()?; let ema_period = env_usize("EMA_PERIOD", "20")?; - let mt5 = Arc::new(mt5_client::Mt5Client::new(mt5_base_url)); + let mt5 = Arc::new(mt5_client::Mt5Client::new(mt5_base_url.clone())); // ── live mode ───────────────────────────────────────────────────────────── let live_mode = std::env::var("LIVE").map(|v| v == "true" || v == "1").unwrap_or(false); if live_mode { let poll_secs = env_u64("LIVE_POLL_SECS", "30")?; + let tg_token = std::env::var("TELEGRAM_BOT_TOKEN").ok().filter(|s| !s.is_empty()); + let tg_chat_id = std::env::var("TELEGRAM_CHAT_ID").ok() + .and_then(|s| s.parse::().ok()); + let tg_thread_id = std::env::var("TELEGRAM_THREAD_ID").ok() + .and_then(|s| s.parse::().ok()); + let telegram = match (tg_token, tg_chat_id) { + (Some(token), Some(chat_id)) => Some(telegram::TelegramConfig { + token, + chat_id, + thread_id: tg_thread_id, + }), + _ => None, + }; + let base_cfg = live::LiveConfig { symbol: String::new(), // filled per-spawn timeframe, @@ -111,6 +126,8 @@ async fn main() -> anyhow::Result<()> { spread_override, ema_period, poll_secs, + mt5_base_url: mt5_base_url.clone(), + telegram, }; let mut handles = Vec::new(); diff --git a/src/telegram.rs b/src/telegram.rs new file mode 100644 index 0000000..1d54114 --- /dev/null +++ b/src/telegram.rs @@ -0,0 +1,67 @@ +use anyhow::{Context, Result}; +use serde::Deserialize; + +#[derive(Debug, Clone)] +pub struct TelegramConfig { + pub token: String, + pub chat_id: i64, + pub thread_id: Option, +} + +#[derive(Deserialize)] +struct TgResp { + ok: bool, + result: Option, + description: Option, +} + +#[derive(Deserialize)] +struct TgMsg { + message_id: i64, +} + +impl TelegramConfig { + fn url(&self, method: &str) -> String { + format!("https://api.telegram.org/bot{}/{}", self.token, method) + } + + pub async fn send(&self, http: &reqwest::Client, text: &str) -> Result { + let mut body = serde_json::json!({ + "chat_id": self.chat_id, + "text": text, + "parse_mode": "HTML", + "disable_web_page_preview": true, + }); + if let Some(tid) = self.thread_id { + body["message_thread_id"] = serde_json::json!(tid); + } + let resp: TgResp = http + .post(self.url("sendMessage")) + .json(&body) + .send().await? + .json().await?; + if !resp.ok { + anyhow::bail!("sendMessage: {}", resp.description.unwrap_or_default()); + } + Ok(resp.result.context("no result")?.message_id) + } + + pub async fn edit(&self, http: &reqwest::Client, msg_id: i64, text: &str) -> Result<()> { + let body = serde_json::json!({ + "chat_id": self.chat_id, + "message_id": msg_id, + "text": text, + "parse_mode": "HTML", + "disable_web_page_preview": true, + }); + let resp: TgResp = http + .post(self.url("editMessageText")) + .json(&body) + .send().await? + .json().await?; + if !resp.ok { + tracing::warn!("Telegram editMessageText: {}", resp.description.unwrap_or_default()); + } + Ok(()) + } +}