MT5Service - Streaming Methods (Mid-Level API)¶

5 methods for real-time streams: ticks, trade events, position profits, ticket changes

🧩 API Layer: MID-LEVEL - wrappers over MT5Account with clean Go types

⚠️ Important Note About Streaming Methods¶

Key insight:

Among the 5 streaming methods, only StreamTicks() provides significant abstraction over the low-level API by converting protobuf timestamps to time.Time.

The other 4 methods (StreamTradeUpdates, StreamPositionProfits, StreamOpenedTickets, StreamTransactions) are thin wrappers that directly pass through channels from MT5Account.

What this means:¶

StreamTicks: Automatically converts Timestamp → time.Time (real convenience)
Other 4 methods: Essentially identical to low-level, just wrapped for consistency
Choice is yours: Using mid-level vs low-level for streaming is largely a matter of preference and API consistency in your codebase

💡 If you prefer working directly with MT5Account for streaming (except ticks), that's perfectly valid!

🎯 Why These Methods Exist¶

Problem: Low-Level Complexity¶

In MT5Account, streaming returns protobuf channels with nested structures:

// Low-level (MT5Account)
dataCh, errCh := account.OnSymbolTick(ctx, &pb.OnSymbolTickRequest{SymbolNames: []string{"EURUSD"}})
for data := range dataCh {
    tick := data.SymbolTick  // ← nested structure
    t := tick.Time.AsTime()   // ← manual time conversion
    fmt.Printf("Tick: %.5f\n", tick.Bid)
}

Solution: Clean Go Types¶

MT5Service automatically converts data and returns clean Go types:

// Mid-level (MT5Service)
tickCh, errCh := service.StreamTicks(ctx, []string{"EURUSD"})
for tick := range tickCh {
    // tick.Time is already time.Time! (broker server time; tick.TimeUtc = UTC)
    fmt.Printf("%s tick at %s: %.5f\n", tick.Symbol, tick.Time.Format("15:04:05"), tick.Bid)
}

Advantages:¶

✅ Automatic Timestamp → time.Time conversion (StreamTicks only), broker time and UTC
✅ Clean *SymbolTick DTO instead of nested protobuf structures (with the tick's Symbol)
✅ Stream messages without a tick are skipped; cancelling ctx always ends the stream, even when you stopped reading the tick channel
✅ Convenient channel types (<-chan *SymbolTick instead of <-chan *pb.OnSymbolTickData)
✅ Clearer method names (Stream* instead of On*)
✅ Automatic unpacking of nested structures

📋 All 5 Methods¶

Method	Returns	Low-Level Equivalent
`StreamTicks(ctx, symbols)`	`(<-chan *SymbolTick, <-chan error)`	`OnSymbolTick(ctx, req)` + time.Time conversion
`StreamTradeUpdates(ctx)`	`(<-chan *pb.OnTradeData, <-chan error)`	`OnTrade(ctx, req)` (pass-through)
`StreamPositionProfits(ctx)`	`(<-chan *pb.OnPositionProfitData, <-chan error)`	`OnPositionProfit(ctx, req)` (pass-through)
`StreamOpenedTickets(ctx)`	`(<-chan *pb.OnPositionsAndPendingOrdersTicketsData, <-chan error)`	`OnPositionsAndPendingOrdersTickets(ctx, req)` (pass-through)
`StreamTransactions(ctx)`	`(<-chan *pb.OnTradeTransactionData, <-chan error)`	`OnTradeTransaction(ctx, req)` (pass-through)

📦 DTO Structures¶

SymbolTick (for StreamTicks)¶

type SymbolTick struct {
    Symbol     string    // Symbol of the tick
    Time       time.Time // Tick time in the broker server's time zone (as MT5 reports it)
    TimeUtc    time.Time // The same moment in UTC
    Bid        float64   // Current Bid price
    Ask        float64   // Current Ask price
    Last       float64   // Last deal price
    Volume     uint64    // Tick volume
    TimeMS     int64     // Tick time in milliseconds (broker server time)
    TimeMSUtc  int64     // Tick time in milliseconds (UTC)
    Flags      uint32    // Tick flags
    VolumeReal float64   // Tick volume with decimal precision
}

Advantage: Time is already time.Time, no manual conversion needed. Time carries the broker's wall clock (its Location() is UTC only as a carrier, so Format prints the broker time); TimeUtc is the real moment.

📖 Method Signatures¶

1) StreamTicks¶

func (s *MT5Service) StreamTicks(
    ctx context.Context,
    symbols []string,
) (<-chan *SymbolTick, <-chan error)

Real-time tick stream for specified symbols.

ADVANTAGE over MT5Account.OnSymbolTick:¶

Returns channel of *SymbolTick structs (clean Go types) with Symbol, broker-time Time/TimeMS and UTC TimeUtc/TimeMSUtc
Automatically converts protobuf OnSymbolTickData → SymbolTick; messages without a tick are skipped
Time is already time.Time (no need for .AsTime())
Cleaner API with separate tick and error channels
Cancelling ctx ends the stream (ctx.Err() on the error channel, then both channels close) even when the consumer stopped reading ticks - the goroutine never blocks forever on a full tick channel

Parameters:¶

symbols - list of symbols to stream (e.g., []string{"EURUSD", "GBPUSD"})

Returns:¶

Read-only channel of *SymbolTick structs
Read-only channel of errors

Note: Channels close when streaming stops.

Example:¶

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

tickCh, errCh := service.StreamTicks(ctx, []string{"EURUSD", "GBPUSD"})

for {
    select {
    case tick, ok := <-tickCh:
        if !ok {
            return  // channel closed
        }
        fmt.Printf("[%s broker / %s UTC] %s Bid=%.5f, Ask=%.5f\n",
            tick.Time.Format("15:04:05"), tick.TimeUtc.Format("15:04:05"),
            tick.Symbol, tick.Bid, tick.Ask)

    case err, ok := <-errCh:
        if !ok {
            return  // channel closed
        }
        fmt.Printf("❌ Error: %v\n", err)
        return
    }
}

⚠️ IMPORTANT: Always read from both channels in a select statement.

2) StreamTradeUpdates¶

func (s *MT5Service) StreamTradeUpdates(
    ctx context.Context,
) (<-chan *pb.OnTradeData, <-chan error)

Real-time stream of trade events (new/disappeared orders and positions, history updates).

Events include:¶

New orders and positions opened
Orders and positions closed/cancelled
Historical orders and deals

Returns:¶

Read-only channel of *pb.OnTradeData (protobuf)
Read-only channel of errors

Example:¶

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

tradeCh, errCh := service.StreamTradeUpdates(ctx)

for {
    select {
    case data, ok := <-tradeCh:
        if !ok {
            return
        }

        ev := data.EventData
        if ev == nil {
            continue
        }

        // New positions
        for _, pos := range ev.NewPositions {
            fmt.Printf("✅ New position #%d: %s, Volume: %.2f\n",
                pos.Ticket, pos.Symbol, pos.Volume)
        }

        // Closed positions
        for _, pos := range ev.DisappearedPositions {
            fmt.Printf("❌ Position closed: #%d\n", pos.Ticket)
        }

        // New deals
        for _, deal := range ev.NewHistoryDeals {
            fmt.Printf("💰 New deal #%d: %s, Profit: %.2f\n",
                deal.Ticket, deal.Symbol, deal.Profit)
        }

    case err, ok := <-errCh:
        if !ok {
            return
        }
        fmt.Printf("❌ Error: %v\n", err)
        return
    }
}

3) StreamPositionProfits¶

func (s *MT5Service) StreamPositionProfits(
    ctx context.Context,
) (<-chan *pb.OnPositionProfitData, <-chan error)

Real-time stream of profit/loss updates for open positions.

Events include:¶

NewPositions - positions opened
UpdatedPositions - positions whose profit changed as prices move
DeletedPositions - positions closed
AccountInfo - account totals (Balance, Equity, Margin, FreeMargin, Profit, MarginLevel)

Each position entry (*pb.OnPositionProfitPositionInfo) has Ticket, Profit and PositionSymbol.

Returns:¶

Read-only channel of *pb.OnPositionProfitData (protobuf)
Read-only channel of errors

Example:¶

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

profitCh, errCh := service.StreamPositionProfits(ctx)

for {
    select {
    case data, ok := <-profitCh:
        if !ok {
            return
        }

        for _, pos := range data.UpdatedPositions {
            fmt.Printf("Position #%d: %s, Profit: %.2f\n",
                pos.Ticket, pos.PositionSymbol, pos.Profit)
        }

        if data.AccountInfo != nil {
            fmt.Printf("📊 Total P&L: %.2f\n", data.AccountInfo.Profit)
        }

    case err, ok := <-errCh:
        if !ok {
            return
        }
        fmt.Printf("❌ Error: %v\n", err)
        return
    }
}

4) StreamOpenedTickets¶

func (s *MT5Service) StreamOpenedTickets(
    ctx context.Context,
) (<-chan *pb.OnPositionsAndPendingOrdersTicketsData, <-chan error)

Real-time stream of updates to open position and pending order ticket numbers.

Events include:¶

List of current position tickets
List of current pending order tickets
Updates when tickets are added/removed

💡 Lightweight alternative to StreamTradeUpdates when you only need ticket IDs.

Returns:¶

Read-only channel of *pb.OnPositionsAndPendingOrdersTicketsData (protobuf)
Read-only channel of errors

Example:¶

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

ticketCh, errCh := service.StreamOpenedTickets(ctx)

for {
    select {
    case data, ok := <-ticketCh:
        if !ok {
            return
        }

        fmt.Printf("Open positions: %d tickets\n",
            len(data.PositionTickets))
        fmt.Printf("  Tickets: %v\n", data.PositionTickets)

        fmt.Printf("Pending orders: %d tickets\n",
            len(data.PendingOrderTickets))
        fmt.Printf("  Tickets: %v\n", data.PendingOrderTickets)

    case err, ok := <-errCh:
        if !ok {
            return
        }
        fmt.Printf("❌ Error: %v\n", err)
        return
    }
}

5) StreamTransactions¶

func (s *MT5Service) StreamTransactions(
    ctx context.Context,
) (<-chan *pb.OnTradeTransactionData, <-chan error)

Real-time stream of all trade transactions (most detailed streaming method).

Events include:¶

Order placement, modification, deletion
Deal execution
Position opening, modification, closing
All intermediate states and changes

🔥 Most powerful streaming method with detailed transaction information.

When to use:¶

Use StreamTradeUpdates for simple trade monitoring
Use StreamPositionProfits for P&L tracking
Use StreamTransactions for detailed analysis of all events

Returns:¶

Read-only channel of *pb.OnTradeTransactionData (protobuf)
Read-only channel of errors

Example:¶

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

txCh, errCh := service.StreamTransactions(ctx)

for {
    select {
    case data, ok := <-txCh:
        if !ok {
            return
        }

        tx := data.TradeTransaction // also: data.TradeRequest, data.TradeResult
        if tx == nil {
            continue
        }

        fmt.Printf("Transaction type: %v\n", tx.Type)
        if tx.DealTicket != 0 {
            fmt.Printf("  Deal #%d executed\n", tx.DealTicket)
        }
        if tx.OrderTicket != 0 {
            fmt.Printf("  Order #%d: %v\n", tx.OrderTicket, tx.OrderType)
        }
        if tx.PositionTicket != 0 {
            fmt.Printf("  Position #%d affected\n", tx.PositionTicket)
        }

    case err, ok := <-errCh:
        if !ok {
            return
        }
        fmt.Printf("❌ Error: %v\n", err)
        return
    }
}

💡 Usage Examples¶

Example 1: Real-time Price Monitoring¶

// ✅ MT5Service - clean types and auto-conversion
ctx := context.Background()

tickCh, errCh := service.StreamTicks(ctx, []string{"EURUSD"})

for {
    select {
    case tick, ok := <-tickCh:
        if !ok {
            return  // channel closed
        }
        // tick.Time is already time.Time! (broker server time)
        fmt.Printf("[%s] Bid: %.5f\n",
            tick.Time.Format("15:04:05"), tick.Bid)

    case err, ok := <-errCh:
        if !ok {
            return
        }
        return err
    }
}

Advantage: time.Time conversion is automatic, no need for .AsTime() calls.

Example 2: Monitor Multiple Symbols¶

// ✅ MT5Service - monitor 3 symbols simultaneously
func MonitorMultipleSymbols(service *mt5.MT5Service) {
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
    defer cancel()

    symbols := []string{"EURUSD", "GBPUSD", "USDJPY"}
    tickCh, errCh := service.StreamTicks(ctx, symbols)

    for {
        select {
        case tick, ok := <-tickCh:
            if !ok {
                fmt.Println("Stream closed")
                return
            }

            // tick.Symbol tells which of the symbols ticked
            fmt.Printf("[%s] %s Tick: Bid=%.5f, Ask=%.5f, Spread=%.5f (price units)\n",
                tick.Time.Format("15:04:05"), tick.Symbol,
                tick.Bid, tick.Ask, tick.Ask-tick.Bid)

        case err, ok := <-errCh:
            if !ok {
                return
            }
            fmt.Printf("❌ Error: %v\n", err)
            return

        case <-ctx.Done():
            fmt.Println("Timeout reached")
            return
        }
    }
}

Example 3: Real-time Total Profit Calculation¶

// ✅ MT5Service - automatic P&L recalculation
func MonitorTotalProfit(service *mt5.MT5Service) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    profitCh, errCh := service.StreamPositionProfits(ctx)

    fmt.Println("📊 Real-time P&L Monitor")
    fmt.Println("Press Ctrl+C to stop...")

    for {
        select {
        case data, ok := <-profitCh:
            if !ok {
                return
            }

            if data.AccountInfo == nil {
                continue
            }
            // Account-wide floating P&L; UpdatedPositions lists only the positions that changed
            totalProfit := data.AccountInfo.Profit
            posCount := len(data.UpdatedPositions)

            // Colored output
            profitColor := "🟢"
            if totalProfit < 0 {
                profitColor = "🔴"
            }

            fmt.Printf("\r%s Updated positions: %d | Total P&L: %.2f   ",
                profitColor, posCount, totalProfit)

        case err, ok := <-errCh:
            if !ok {
                return
            }
            fmt.Printf("\n❌ Error: %v\n", err)
            return
        }
    }
}

Example 4: Trade Event Notifications¶

// ✅ MT5Service - trade event notifications
func NotifyTradeEvents(service *mt5.MT5Service) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    tradeCh, errCh := service.StreamTradeUpdates(ctx)

    fmt.Println("🔔 Trade Event Monitor")

    for {
        select {
        case data, ok := <-tradeCh:
            if !ok {
                return
            }

            ev := data.EventData
            if ev == nil {
                continue
            }

            // New positions
            for _, pos := range ev.NewPositions {
                typeStr := "BUY"
                if pos.Type == pb.SUB_ENUM_POSITION_TYPE_SUB_POSITION_TYPE_SELL {
                    typeStr = "SELL"
                }
                fmt.Printf("\n🆕 Position opened:\n")
                fmt.Printf("   #%d: %s %s %.2f lots @ %.5f\n",
                    pos.Ticket, typeStr, pos.Symbol, pos.Volume, pos.PriceOpen)
            }

            // Closed positions
            for _, pos := range ev.DisappearedPositions {
                fmt.Printf("\n🔒 Position closed: #%d\n", pos.Ticket)
            }

            // New deals
            for _, deal := range ev.NewHistoryDeals {
                profitStr := "Profit"
                if deal.Profit < 0 {
                    profitStr = "Loss"
                }
                fmt.Printf("\n�� Deal executed:\n")
                fmt.Printf("   #%d: %s, %s: %.2f\n",
                    deal.Ticket, deal.Symbol, profitStr, deal.Profit)
            }

        case err, ok := <-errCh:
            if !ok {
                return
            }
            fmt.Printf("\n❌ Error: %v\n", err)
            return
        }
    }
}

Example 5: Log All Transactions¶

// ✅ MT5Service - detailed transaction logging
func LogAllTransactions(service *mt5.MT5Service, logFile string) error {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    f, err := os.Create(logFile)
    if err != nil {
        return err
    }
    defer f.Close()

    txCh, errCh := service.StreamTransactions(ctx)

    fmt.Println("📝 Transaction Logger started")
    fmt.Printf("Logging to: %s\n", logFile)

    for {
        select {
        case data, ok := <-txCh:
            if !ok {
                return nil
            }

            tx := data.TradeTransaction
            if tx == nil {
                continue
            }

            timestamp := time.Now().Format("2006-01-02 15:04:05")
            logLine := fmt.Sprintf("[%s] Type: %v, Order: %d, Deal: %d, Position: %d\n",
                timestamp, tx.Type, tx.OrderTicket, tx.DealTicket, tx.PositionTicket)

            // Write to file
            f.WriteString(logLine)
            f.Sync()  // Flush buffer

            // Console output
            fmt.Print(logLine)

        case err, ok := <-errCh:
            if !ok {
                return nil
            }
            return err
        }
    }
}

🔧 When to Use Each Method¶

✅ StreamTicks¶

Use when:¶

Need real-time price monitoring
Building charts or indicators
Tracking spread in real-time
Implementing algorithmic trading

Example:¶

tickCh, errCh := service.StreamTicks(ctx, []string{"EURUSD"})
for tick := range tickCh {
    if tick.Ask - tick.Bid > maxSpread {
        // Wide spread, don't trade
    }
}

✅ StreamTradeUpdates¶

Use when:¶

Monitoring new/closed positions
Tracking trading activity
Need notifications about deals
Building trade journal

Example:¶

tradeCh, errCh := service.StreamTradeUpdates(ctx)
for data := range tradeCh {
    if data.EventData == nil {
        continue
    }
    for _, pos := range data.EventData.NewPositions {
        // Notification about new position
    }
}

✅ StreamPositionProfits¶

Use when:¶

Tracking P&L in real-time
Building profit dashboard
Implementing automatic closing on profit target
Monitoring risk

Example:¶

profitCh, errCh := service.StreamPositionProfits(ctx)
for data := range profitCh {
    if data.AccountInfo == nil {
        continue
    }
    totalProfit := data.AccountInfo.Profit
    // Check profit/loss limits
}

✅ StreamOpenedTickets¶

Use when:¶

Only need ticket numbers (lightweight)
Monitoring number of positions
Checking existence of specific ticket

Example:¶

ticketCh, errCh := service.StreamOpenedTickets(ctx)
for data := range ticketCh {
    if len(data.PositionTickets) > maxPositions {
        // Too many positions, don't open new ones
    }
}

✅ StreamTransactions¶

Use when:¶

Need detailed information about all events
Building audit log of trading
Tracking intermediate states
Analyzing trade flow

Example:¶

txCh, errCh := service.StreamTransactions(ctx)
for data := range txCh {
    // Log all events
    log.Printf("Transaction: %v", data)
}

📊 Method Comparison¶

Method	Detail Level	Load	Usage
StreamTicks	High (every tick)	High	Algo trading, charts
StreamTradeUpdates	Medium (trade events)	Medium	Position monitoring
StreamPositionProfits	Medium (P&L updates)	Medium	Profit dashboard
StreamOpenedTickets	Low (tickets only)	Low	Lightweight monitoring
StreamTransactions	Maximum (all events)	High	Full audit

💡 Best Practices & Recommendations¶

1. Always use context to manage stream lifetime¶

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

2. Read from both channels in select statement¶

select {
case data := <-dataCh:
    // processing
case err := <-errCh:
    // error handling
}

3. Check channel closure¶

case data, ok := <-dataCh:
    if !ok {
        return  // channel closed
    }

4. Use timeout for long-running streams¶

ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)

5. Choose lightweight methods when possible¶

For lightweight monitoring, use StreamOpenedTickets instead of StreamTradeUpdates

6. Handle errors properly¶

Streams can be interrupted - always handle error channels

🛡️ "Graceful Shutdown" Pattern¶

func StreamWithShutdown(service *mt5.MT5Service) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Handle Ctrl+C
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, os.Interrupt)
    go func() {
        <-sigCh
        fmt.Println("\nShutting down...")
        cancel()
    }()

    tickCh, errCh := service.StreamTicks(ctx, []string{"EURUSD"})

    for {
        select {
        case tick, ok := <-tickCh:
            if !ok {
                return
            }
            // Process tick

        case err, ok := <-errCh:
            if !ok {
                return
            }
            fmt.Printf("Error: %v\n", err)
            return

        case <-ctx.Done():
            fmt.Println("Context cancelled")
            return
        }
    }
}

MT5Service (Mid-Level):¶

MT5Service Overview - general mid-level API overview
Account Methods - account information retrieval
Symbol Methods - symbol parameters and quotes
Position & Orders Methods - query positions and orders
Trading Methods - place and manage orders
Market Depth Methods - order book operations

MT5Account (Low-Level):¶

Streaming Methods - low-level streaming operations
MT5Account Master Overview - complete low-level API reference

🎯 Summary¶

MT5Service Streaming methods solve the main task - simplifying real-time monitoring:

What You Avoid:¶

❌ No need to unpack data.SymbolTick and nested structures
❌ No need to manually convert Timestamp.AsTime() (StreamTicks only)
❌ No need to work with <-chan *pb.OnSymbolTickData

What You Get:¶

✅ Clean Go types (*SymbolTick with time.Time)
✅ Convenient channels (<-chan *SymbolTick, <-chan error)
✅ Automatic time conversion (StreamTicks)
✅ Code reads like standard Go

Real-time monitoring in 3 lines:¶

tickCh, errCh := service.StreamTicks(ctx, []string{"EURUSD"})
for tick := range tickCh {
    fmt.Printf("Price: %.5f at %s\n", tick.Bid, tick.Time)
}

You now have all the tools for real-time market monitoring!

Whether you choose mid-level or low-level streaming APIs, both will serve you well. The choice is yours based on your coding style and project requirements.

Happy trading and may your streams be fast and your profits be plentiful! 📈

MT5Service - Streaming Methods (Mid-Level API)¶

⚠️ Important Note About Streaming Methods¶

What this means:¶

🎯 Why These Methods Exist¶

Problem: Low-Level Complexity¶

Solution: Clean Go Types¶

Advantages:¶

📋 All 5 Methods¶

📦 DTO Structures¶

SymbolTick (for StreamTicks)¶

📖 Method Signatures¶

1) StreamTicks¶

ADVANTAGE over MT5Account.OnSymbolTick:¶

Parameters:¶

Returns:¶

Example:¶

2) StreamTradeUpdates¶

Events include:¶

Returns:¶

Example:¶

3) StreamPositionProfits¶

Events include:¶

Returns:¶

Example:¶

4) StreamOpenedTickets¶

Events include:¶

Returns:¶

Example:¶

5) StreamTransactions¶

Events include:¶

When to use:¶

Returns:¶

Example:¶

💡 Usage Examples¶

Example 1: Real-time Price Monitoring¶

Example 2: Monitor Multiple Symbols¶

Example 3: Real-time Total Profit Calculation¶

Example 4: Trade Event Notifications¶

Example 5: Log All Transactions¶

🔧 When to Use Each Method¶

✅ StreamTicks¶

Use when:¶

Example:¶

✅ StreamTradeUpdates¶

Use when:¶

Example:¶

✅ StreamPositionProfits¶

Use when:¶

Example:¶

✅ StreamOpenedTickets¶

Use when:¶

Example:¶

✅ StreamTransactions¶

Use when:¶

Example:¶

📊 Method Comparison¶

💡 Best Practices & Recommendations¶

1. Always use context to manage stream lifetime¶

2. Read from both channels in select statement¶

3. Check channel closure¶

4. Use timeout for long-running streams¶

5. Choose lightweight methods when possible¶

6. Handle errors properly¶

🛡️ "Graceful Shutdown" Pattern¶

📚 Related Sections¶

MT5Service (Mid-Level):¶

MT5Account (Low-Level):¶

🎯 Summary¶

What You Avoid:¶

What You Get:¶

Real-time monitoring in 3 lines:¶