PineSript/Buyer/BitgetWebSocketClient.cs

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using PineScriptV6;
using Serilog;

namespace Buyer
{
    /// <summary>
    /// Bitget Public WebSocket — 실시간 캔들 스트림 수신.
    /// confirm="1" 인 완성 봉만 OnCandle 이벤트로 발행합니다.
    /// 연결 끊김 시 자동 재연결합니다.
    /// </summary>
    public class BitgetWebSocketClient : IAsyncDisposable
    {
        private readonly BitgetConfig _cfg;

        /// <summary>완성된 봉(confirm=1)이 도착했을 때 발생합니다.</summary>
        public event Action<CandleData>? OnCandle;

        /// <summary>WebSocket 연결/재연결 상태 변경 시 발생합니다.</summary>
        public event Action<bool>? OnConnectionChanged;

        private CancellationTokenSource _cts = new();

        public BitgetWebSocketClient(BitgetConfig cfg) => _cfg = cfg;

        /// <summary>
        /// 연결을 시작하고 수신 루프를 백그라운드로 실행합니다.
        /// </summary>
        public Task StartAsync(CancellationToken ct)
        {
            _cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
            _ = Task.Run(() => RunLoop(_cts.Token), _cts.Token);
            return Task.CompletedTask;
        }

        // ════════════════════════════════════════════════════════════════
        // 내부 — 연결 루프 (재연결 포함)
        // ════════════════════════════════════════════════════════════════

        private async Task RunLoop(CancellationToken ct)
        {
            while (!ct.IsCancellationRequested)
            {
                using var ws = new ClientWebSocket();
                try
                {
                    await ws.ConnectAsync(new Uri(_cfg.WsUrl), ct);
                    Log.Information("[WS] 연결됨 → {Url}", _cfg.WsUrl);
                    OnConnectionChanged?.Invoke(true);

                    await SubscribeAsync(ws, ct);
                    await Task.WhenAll(
                        PingLoopAsync(ws, ct),
                        ReceiveLoopAsync(ws, ct));
                }
                catch (OperationCanceledException) when (ct.IsCancellationRequested)
                {
                    break;  // 정상 종료
                }
                catch (Exception ex)
                {
                    Log.Warning(ex, "[WS] 연결 오류 — {Delay}초 후 재연결",
                        _cfg.WsReconnectDelaySec);
                }

                OnConnectionChanged?.Invoke(false);

                if (!ct.IsCancellationRequested)
                    await Task.Delay(_cfg.WsReconnectDelaySec * 1000, ct)
                              .ConfigureAwait(false);
            }
            Log.Information("[WS] 연결 루프 종료");
        }

        // ── 구독 요청 ─────────────────────────────────────────────────

        private async Task SubscribeAsync(ClientWebSocket ws, CancellationToken ct)
        {
            var payload = new
            {
                op   = "subscribe",
                args = new[]
                {
                    new
                    {
                        instType = "USDT-FUTURES",
                        channel  = _cfg.WsChannel,
                        instId   = _cfg.Symbol
                    }
                }
            };

            string json = JsonSerializer.Serialize(payload);
            await SendAsync(ws, json, ct);
            Log.Information("[WS] 구독 요청: {Channel} {Symbol}",
                _cfg.WsChannel, _cfg.Symbol);
        }

        // ── Ping 루프 ────────────────────────────────────────────────

        private async Task PingLoopAsync(ClientWebSocket ws, CancellationToken ct)
        {
            while (ws.State == WebSocketState.Open && !ct.IsCancellationRequested)
            {
                await Task.Delay(_cfg.WsPingInterval * 1000, ct);
                if (ws.State == WebSocketState.Open)
                {
                    await SendAsync(ws, "ping", ct);
                    Log.Debug("[WS] ping 전송");
                }
            }
        }

        // ── 수신 루프 ────────────────────────────────────────────────

        private async Task ReceiveLoopAsync(ClientWebSocket ws, CancellationToken ct)
        {
            var buf = new byte[65536];
            var sb  = new StringBuilder();

            while (ws.State == WebSocketState.Open && !ct.IsCancellationRequested)
            {
                WebSocketReceiveResult result;
                try
                {
                    result = await ws.ReceiveAsync(new ArraySegment<byte>(buf), ct);
                }
                catch (OperationCanceledException) { break; }
                catch (WebSocketException ex)
                {
                    Log.Warning("[WS] 수신 오류: {Msg}", ex.Message);
                    break;
                }

                if (result.MessageType == WebSocketMessageType.Close)
                {
                    Log.Warning("[WS] 서버 Close 프레임 수신");
                    break;
                }

                sb.Append(Encoding.UTF8.GetString(buf, 0, result.Count));

                if (!result.EndOfMessage) continue;

                string msg = sb.ToString();
                sb.Clear();

                if (msg == "pong") { Log.Debug("[WS] pong 수신"); continue; }
                ProcessMessage(msg);
            }
        }

        // ── 메시지 처리 ──────────────────────────────────────────────

        private void ProcessMessage(string msg)
        {
            try
            {
                using var doc  = JsonDocument.Parse(msg);
                var       root = doc.RootElement;

                // 이벤트 응답 (subscribe 확인 등) 무시
                if (root.TryGetProperty("event", out _))
                {
                    Log.Debug("[WS] 이벤트: {Msg}", msg);
                    return;
                }

                if (!root.TryGetProperty("data", out var dataEl)) return;

                foreach (var item in dataEl.EnumerateArray())
                {
                    var arr = item.EnumerateArray().ToList();
                    // Bitget candle 배열: [ts, o, h, l, c, baseVol, quoteVol, usdtVol, confirm]
                    if (arr.Count < 6) continue;

                    // confirm: 마지막 원소 "1" = 완성 봉
                    string confirm = arr[^1].GetString() ?? "0";
                    if (confirm != "1")
                    {
                        Log.Debug("[WS] 미완성 봉 수신 (무시): {Ts}", arr[0].GetString());
                        continue;
                    }

                    long    ts  = long.Parse(arr[0].GetString()!);
                    decimal o   = decimal.Parse(arr[1].GetString()!,
                                    System.Globalization.CultureInfo.InvariantCulture);
                    decimal h   = decimal.Parse(arr[2].GetString()!,
                                    System.Globalization.CultureInfo.InvariantCulture);
                    decimal l   = decimal.Parse(arr[3].GetString()!,
                                    System.Globalization.CultureInfo.InvariantCulture);
                    decimal c   = decimal.Parse(arr[4].GetString()!,
                                    System.Globalization.CultureInfo.InvariantCulture);
                    decimal vol = decimal.Parse(arr[5].GetString()!,
                                    System.Globalization.CultureInfo.InvariantCulture);

                    var candle = new CandleData
                    {
                        Time   = DateTimeOffset.FromUnixTimeMilliseconds(ts),
                        Open   = o,
                        High   = h,
                        Low    = l,
                        Close  = c,
                        Volume = vol
                    };

                    Log.Information("[WS] ✅ 완성 봉: {Time}  O={O} H={H} L={L} C={C} V={V}",
                        candle.Time.ToString("yyyy-MM-dd HH:mm"),
                        candle.Open, candle.High, candle.Low, candle.Close, candle.Volume);

                    OnCandle?.Invoke(candle);
                }
            }
            catch (Exception ex)
            {
                Log.Debug("[WS] 메시지 파싱 오류: {Err} | Raw={Msg}", ex.Message, msg);
            }
        }

        // ── 전송 헬퍼 ────────────────────────────────────────────────

        private static async Task SendAsync(ClientWebSocket ws, string text, CancellationToken ct)
        {
            if (ws.State != WebSocketState.Open) return;
            byte[] data = Encoding.UTF8.GetBytes(text);
            await ws.SendAsync(new ArraySegment<byte>(data),
                WebSocketMessageType.Text, true, ct);
        }

        public ValueTask DisposeAsync()
        {
            _cts.Cancel();
            _cts.Dispose();
            return ValueTask.CompletedTask;
        }
    }
}