如何在Flutter中使用Upstox WebSocketChannel获取实时股票数据

原HTTP实现代码 #

class UpstoxService {
  final String accessToken = "token";
  final JsonService jsonService;

  UpstoxService(this.jsonService);


    Future<String?> getInstrumentKey(String tradingSymbol) async {
    try {
      final List<dynamic> data = await jsonService.loadJsonData();
      for (var item in data) {
        if (item['trading_symbol'] == tradingSymbol) {
          return item['instrument_key'];
        }
      }
      return null; // 返回null表示未找到匹配项
    } catch (e) {
      print("查找instrument key出错: $e");
      return null;
    }
  }


  Future<Map<String, String>> fetchStockData(String instrumentKey,String symbol) async {
    final url = Uri.parse('https://api.upstox.com/v2/market-quote/quotes?instrument_key=$instrumentKey');
    final headers = {
      'Accept': 'application/json',
      'Authorization': 'Bearer $accessToken',
    };

    final response = await http.get(url, headers: headers);
    if (response.statusCode == 200) {
      var data = jsonDecode(response.body);
      Map<String,String> extractData =  formatData(data['data']["NSE_EQ:$symbol"]);
      return extractData;
    } else {
      throw Exception('无可用数据');
    }
  }

  Map<String, String> formatData(var data) {
    double open = data['ohlc']!['open'];
    String currentPrice = data['last_price'].toString(); // 修正：实时数据取last_price而非close
    double netChange = data['net_change'];
    String percentageChange = ((netChange/open)*100).toStringAsFixed(2);

    return  {
      "currentPrice": currentPrice,
      "percentageChange":"$percentageChange%",
      "amountChange":netChange.toString(),
    };
  }
}

WebSocket实现方案 #

1. 添加依赖 #

先在项目中引入web_socket_channel包：

flutter pub add web_socket_channel

2. 重构UpstoxService实现WebSocket逻辑 #

import 'package:web_socket_channel/web_socket_channel.dart';
import 'dart:convert';

class UpstoxService {
  final String accessToken = "token";
  final JsonService jsonService;
  WebSocketChannel? _channel;
  Stream<Map<String, String>>? _stockDataStream;

  UpstoxService(this.jsonService);

  // 保留原有工具方法
  Future<String?> getInstrumentKey(String tradingSymbol) async {
    try {
      final List<dynamic> data = await jsonService.loadJsonData();
      for (var item in data) {
        if (item['trading_symbol'] == tradingSymbol) {
          return item['instrument_key'];
        }
      }
      return null;
    } catch (e) {
      print("查找instrument key出错: $e");
      return null;
    }
  }

  Map<String, String> formatData(var data) {
    double open = data['ohlc']!['open'];
    String currentPrice = data['last_price'].toString();
    double netChange = data['net_change'];
    String percentageChange = ((netChange / open) * 100).toStringAsFixed(2);

    return {
      "currentPrice": currentPrice,
      "percentageChange": "$percentageChange%",
      "amountChange": netChange.toString(),
    };
  }

  // 建立WebSocket连接并订阅指定股票
  Future<void> connectAndSubscribe(String instrumentKey) async {
    // Upstox WebSocket官方端点，需替换为实际API Key
    final wsUrl = Uri.parse('wss://api.upstox.com/v2/feed/market-data-stream?apiKey=YOUR_API_KEY&accessToken=$accessToken');
    
    _channel = WebSocketChannel.connect(wsUrl);

    // 发送订阅指令，格式遵循Upstox API文档要求
    final subscribeMessage = jsonEncode({
      "type": "subscribe",
      "instrumentKeys": [instrumentKey]
    });
    _channel?.sink.add(subscribeMessage);

    // 将WebSocket消息转换为格式化后的数据流
    _stockDataStream = _channel?.stream.map((message) {
      final data = jsonDecode(message);
      return formatData(data['data'][instrumentKey]);
    });
  }

  // 对外暴露实时数据流，供UI监听
  Stream<Map<String, String>>? get stockDataStream => _stockDataStream;

  // 关闭WebSocket连接
  void disconnect() {
    _channel?.sink.close();
    _channel = null;
    _stockDataStream = null;
  }
}

3. 在Widget中监听实时数据 #

class StockPriceWidget extends StatefulWidget {
  final String tradingSymbol;
  const StockPriceWidget({super.key, required this.tradingSymbol});

  @override
  State<StockPriceWidget> createState() => _StockPriceWidgetState();
}

class _StockPriceWidgetState extends State<StockPriceWidget> {
  late UpstoxService _upstoxService;
  Map<String, String>? _currentStockData;

  @override
  void initState() {
    super.initState();
    _upstoxService = UpstoxService(JsonService()); // 假设JsonService已实现
    _initWebSocket();
  }

  Future<void> _initWebSocket() async {
    final instrumentKey = await _upstoxService.getInstrumentKey(widget.tradingSymbol);
    if (instrumentKey != null) {
      await _upstoxService.connectAndSubscribe(instrumentKey);
      _upstoxService.stockDataStream?.listen((data) {
        setState(() {
          _currentStockData = data;
        });
      });
    }
  }

  @override
  void dispose() {
    _upstoxService.disconnect();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return _currentStockData == null
        ? const CircularProgressIndicator()
        : Column(
            children: [
              Text("当前价格: ${_currentStockData!['currentPrice']}"),
              Text("涨跌幅: ${_currentStockData!['percentageChange']}"),
              Text("涨跌额: ${_currentStockData!['amountChange']}"),
            ],
          );
  }
}

注意事项 #

• 确保WebSocket端点和订阅消息格式完全匹配Upstox官方API文档，上述代码中的示例需根据实际文档调整
• 需添加连接异常处理逻辑，避免因网络波动导致应用崩溃
• 注意accessToken的有效期，过期后需要重新获取并重建WebSocket连接

内容的提问来源于stack exchange，提问作者zen