using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using PineScriptV6;
using Serilog;
namespace Buyer
{
///
/// Bitget Public WebSocket — 실시간 캔들 스트림 수신.
/// confirm="1" 인 완성 봉만 OnCandle 이벤트로 발행합니다.
/// 연결 끊김 시 자동 재연결합니다.
///
public class BitgetWebSocketClient : IAsyncDisposable
{
private readonly BitgetConfig _cfg;
/// 완성된 봉(confirm=1)이 도착했을 때 발생합니다.
public event Action? OnCandle;
/// WebSocket 연결/재연결 상태 변경 시 발생합니다.
public event Action? OnConnectionChanged;
private CancellationTokenSource _cts = new();
public BitgetWebSocketClient(BitgetConfig cfg) => _cfg = cfg;
///
/// 연결을 시작하고 수신 루프를 백그라운드로 실행합니다.
///
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(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(data),
WebSocketMessageType.Text, true, ct);
}
public ValueTask DisposeAsync()
{
_cts.Cancel();
_cts.Dispose();
return ValueTask.CompletedTask;
}
}
}