diff --git a/flutter_monisuo/lib/data/models/kline_candle.dart b/flutter_monisuo/lib/data/models/kline_candle.dart new file mode 100644 index 0000000..9b653be --- /dev/null +++ b/flutter_monisuo/lib/data/models/kline_candle.dart @@ -0,0 +1,88 @@ +/// K线蜡烛数据模型 +class KlineCandle { + final String coinCode; + final String interval; + final int openTime; + final double openPrice; + final double highPrice; + final double lowPrice; + final double closePrice; + final double volume; + final int closeTime; + final bool isClosed; + final int? timestamp; + + const KlineCandle({ + required this.coinCode, + required this.interval, + required this.openTime, + required this.openPrice, + required this.highPrice, + required this.lowPrice, + required this.closePrice, + required this.volume, + required this.closeTime, + this.isClosed = true, + this.timestamp, + }); + + factory KlineCandle.fromJson(Map json) { + return KlineCandle( + coinCode: json['coinCode'] as String? ?? '', + interval: json['interval'] as String? ?? '1h', + openTime: json['openTime'] as int? ?? 0, + openPrice: _toDouble(json['openPrice']), + highPrice: _toDouble(json['highPrice']), + lowPrice: _toDouble(json['lowPrice']), + closePrice: _toDouble(json['closePrice']), + volume: _toDouble(json['volume']), + closeTime: json['closeTime'] as int? ?? 0, + isClosed: json['isClosed'] as bool? ?? true, + timestamp: json['timestamp'] as int?, + ); + } + + static double _toDouble(dynamic v) { + if (v == null) return 0.0; + if (v is double) return v; + if (v is int) return v.toDouble(); + if (v is String) return double.tryParse(v) ?? 0.0; + return 0.0; + } + + /// 转换为 k_chart 库的 KLineEntity 格式 + Map toKLineEntityMap() { + return { + 'open': openPrice, + 'high': highPrice, + 'low': lowPrice, + 'close': closePrice, + 'vol': volume, + 'amount': closePrice * volume, + 'time': openTime, + 'id': openTime, + }; + } + + /// 转换为 k_chart KLineEntity 对象 + dynamic toKLineEntity() { + // k_chart KLineEntity.fromCustom 构造器 + return null; // placeholder, actual conversion in kline_page + } + + /// 从 REST API JSON 转换(历史K线) + factory KlineCandle.fromHistoryJson(Map json) { + return KlineCandle( + coinCode: json['coinCode'] as String? ?? '', + interval: json['interval'] as String? ?? '1h', + openTime: json['openTime'] as int? ?? 0, + openPrice: _toDouble(json['openPrice']), + highPrice: _toDouble(json['highPrice']), + lowPrice: _toDouble(json['lowPrice']), + closePrice: _toDouble(json['closePrice']), + volume: _toDouble(json['volume']), + closeTime: json['closeTime'] as int? ?? 0, + isClosed: true, + ); + } +} diff --git a/flutter_monisuo/lib/data/services/kline_service.dart b/flutter_monisuo/lib/data/services/kline_service.dart new file mode 100644 index 0000000..94ad948 --- /dev/null +++ b/flutter_monisuo/lib/data/services/kline_service.dart @@ -0,0 +1,72 @@ +import '../../core/constants/api_endpoints.dart'; +import '../../core/network/api_response.dart'; +import '../../core/network/dio_client.dart'; +import '../models/kline_candle.dart'; + +/// K线 REST API 服务 +class KlineService { + final DioClient _client; + + KlineService(this._client); + + /// 获取历史K线数据 + Future>> fetchHistory({ + required String coinCode, + required String interval, + int limit = 200, + int? before, + }) async { + final params = { + 'coinCode': coinCode, + 'interval': interval, + 'limit': limit, + }; + if (before != null) params['before'] = before; + + final response = await _client.get>( + ApiEndpoints.klineHistory, + queryParameters: params, + ); + + if (response.success && response.data != null) { + final list = response.data!['list'] as List? ?? []; + final candles = list + .map((e) => KlineCandle.fromHistoryJson(e as Map)) + .toList(); + return ApiResponse.success(candles, response.message); + } + return ApiResponse.fail(response.message ?? '获取K线数据失败'); + } + + /// 获取当前进行中的K线 + Future> fetchCurrentCandle({ + required String coinCode, + required String interval, + }) async { + final response = await _client.get>( + ApiEndpoints.klineCurrent, + queryParameters: {'coinCode': coinCode, 'interval': interval}, + ); + + if (response.success && response.data != null) { + return ApiResponse.success( + KlineCandle.fromJson(response.data!), + response.message, + ); + } + return ApiResponse.fail(response.message ?? '获取当前K线失败'); + } + + /// 获取支持的周期列表 + Future>> fetchIntervals() async { + final response = await _client.get>( + ApiEndpoints.klineIntervals, + ); + + if (response.success && response.data != null) { + final list = response.data!['list'] as List? ?? []; + return ApiResponse.success(list.cast(), response.message); + } + return ApiResponse.fail(response.message ?? '获取周期列表失败'); + } +} diff --git a/flutter_monisuo/lib/data/services/kline_websocket_service.dart b/flutter_monisuo/lib/data/services/kline_websocket_service.dart new file mode 100644 index 0000000..bdb201c --- /dev/null +++ b/flutter_monisuo/lib/data/services/kline_websocket_service.dart @@ -0,0 +1,135 @@ +import 'dart:async'; +import 'dart:convert'; +import 'package:stomp_dart_client/stomp_dart_client.dart'; +import '../../core/constants/api_endpoints.dart'; +import '../models/kline_candle.dart'; + +/// K线 WebSocket 服务(STOMP 协议) +class KlineWebSocketService { + StompClient? _stompClient; + final Map _subscriptions = {}; + final Map> _controllers = {}; + + bool _isConnected = false; + bool _isConnecting = false; + int _reconnectDelay = 2000; // 初始重连延迟 + static const int _maxReconnectDelay = 30000; + + /// 订阅某个币种的K线数据 + Stream subscribe(String coinCode) { + final key = coinCode.toUpperCase(); + if (!_controllers.containsKey(key)) { + _controllers[key] = StreamController.broadcast(); + } + _doSubscribe(key); + return _controllers[key]!.stream; + } + + /// 取消订阅 + void unsubscribe(String coinCode) { + final key = coinCode.toUpperCase(); + _subscriptions[key]?.call(); + _subscriptions.remove(key); + } + + /// 连接状态 + bool get isConnected => _isConnected; + + /// 连接 WebSocket + void connect() { + if (_isConnecting || _isConnected) return; + _isConnecting = true; + + final wsUrl = ApiEndpoints.klineWs; + + _stompClient = StompClient( + config: StompConfig( + url: wsUrl, + onConnect: _onConnect, + onDisconnect: _onDisconnect, + onStompError: _onError, + onWebSocketError: _onError, + reconnectDelay: const Duration(milliseconds: 5000), + heartbeatIncoming: const Duration(seconds: 20), + heartbeatOutgoing: const Duration(seconds: 20), + ), + ); + + _stompClient!.activate(); + } + + /// 断开连接 + void disconnect() { + _isConnecting = false; + _stompClient?.deactivate(); + _stompClient = null; + _isConnected = false; + _subscriptions.clear(); + } + + void _onConnect(StompFrame frame) { + _isConnected = true; + _isConnecting = false; + _reconnectDelay = 2000; // 重置重连延迟 + + // 重新订阅所有已注册的币种 + for (final key in _controllers.keys) { + _doSubscribe(key); + } + } + + void _onDisconnect(StompFrame? frame) { + _isConnected = false; + _isConnecting = false; + _scheduleReconnect(); + } + + void _onError(dynamic error) { + _isConnected = false; + _isConnecting = false; + _scheduleReconnect(); + } + + void _scheduleReconnect() { + Future.delayed(Duration(milliseconds: _reconnectDelay), () { + if (!_isConnected && !_isConnecting) { + _reconnectDelay = (_reconnectDelay * 2).clamp(2000, _maxReconnectDelay); + connect(); + } + }); + } + + void _doSubscribe(String coinCode) { + if (_stompClient == null || !_isConnected) { + connect(); // 触发连接,连接成功后会自动重新订阅 + return; + } + + // 避免重复订阅 + if (_subscriptions.containsKey(coinCode)) return; + + final dest = '/topic/kline/$coinCode'; + final sub = _stompClient!.subscribe( + destination: dest, + callback: (StompFrame frame) { + if (frame.body != null) { + try { + final json = jsonDecode(frame.body!) as Map; + final candle = KlineCandle.fromJson(json); + _controllers[coinCode]?.add(candle); + } catch (_) {} + } + }, + ); + _subscriptions[coinCode] = sub; + } + + /// 释放资源 + void dispose() { + disconnect(); + for (final controller in _controllers.values) { + controller.close(); + } + _controllers.clear(); + } +} diff --git a/flutter_monisuo/lib/providers/kline_provider.dart b/flutter_monisuo/lib/providers/kline_provider.dart new file mode 100644 index 0000000..41748c2 --- /dev/null +++ b/flutter_monisuo/lib/providers/kline_provider.dart @@ -0,0 +1,183 @@ +import 'dart:async'; +import 'package:flutter/material.dart'; +import '../data/models/kline_candle.dart'; +import '../data/services/kline_service.dart'; +import '../data/services/kline_websocket_service.dart'; + +/// K线状态管理 +class KlineProvider extends ChangeNotifier { + final KlineService _klineService; + final KlineWebSocketService _wsService; + + KlineProvider(this._klineService, this._wsService); + + List _candles = []; + KlineCandle? _currentCandle; + String _interval = '1h'; + String _coinCode = ''; + bool _isLoading = false; + bool _isLoadingMore = false; + bool _isConnected = false; + String? _error; + + StreamSubscription? _wsSubscription; + Timer? _pollingTimer; + + // Getters + List get candles => _candles; + KlineCandle? get currentCandle => _currentCandle; + String get interval => _interval; + String get coinCode => _coinCode; + bool get isLoading => _isLoading; + bool get isLoadingMore => _isLoadingMore; + bool get isConnected => _isConnected; + String? get error => _error; + + /// 加载某个币种的K线数据 + Future loadCoin(String coinCode, {String? interval}) async { + _coinCode = coinCode; + if (interval != null) _interval = interval; + _candles = []; + _currentCandle = null; + _error = null; + _isLoading = true; + notifyListeners(); + + try { + // 1. 获取历史K线 + final response = await _klineService.fetchHistory( + coinCode: _coinCode, + interval: _interval, + limit: 200, + ); + if (response.success && response.data != null) { + _candles = response.data!; + } + + // 2. 获取当前K线 + final currentResponse = await _klineService.fetchCurrentCandle( + coinCode: _coinCode, + interval: _interval, + ); + if (currentResponse.success && currentResponse.data != null) { + _currentCandle = currentResponse.data; + } + + // 3. 连接 WebSocket + _connectWebSocket(); + + _isLoading = false; + notifyListeners(); + } catch (e) { + _error = '加载K线数据失败: $e'; + _isLoading = false; + notifyListeners(); + } + } + + /// 切换周期 + Future changeInterval(String newInterval) async { + if (newInterval == _interval) return; + + _interval = newInterval; + _candles = []; + _currentCandle = null; + + // 重新加载 + await loadCoin(_coinCode, interval: newInterval); + } + + /// 加载更多历史数据(分页) + Future loadMore() async { + if (_isLoadingMore || _candles.isEmpty) return; + _isLoadingMore = true; + notifyListeners(); + + try { + final oldestTime = _candles.first.closeTime; + final response = await _klineService.fetchHistory( + coinCode: _coinCode, + interval: _interval, + limit: 200, + before: oldestTime, + ); + if (response.success && response.data != null) { + _candles = [...response.data!, ..._candles]; + } + } catch (_) {} + + _isLoadingMore = false; + notifyListeners(); + } + + void _connectWebSocket() { + // 取消之前的订阅 + _wsSubscription?.cancel(); + _wsService.unsubscribe(_coinCode); + + // 订阅新币种 + _wsService.connect(); + _wsSubscription = _wsService.subscribe(_coinCode).listen( + _onTick, + onError: (_) => _startPolling(), + onDone: () => _startPolling(), + ); + + _isConnected = _wsService.isConnected; + notifyListeners(); + } + + void _onTick(KlineCandle tick) { + if (tick.interval != _interval) return; + + _isConnected = true; + + if (tick.isClosed) { + // 收盘 tick → 添加到历史列表 + _candles.add(tick); + _currentCandle = null; + + // 停止轮询(如果之前在轮询) + _pollingTimer?.cancel(); + _pollingTimer = null; + } else { + // 进行中的 tick → 更新当前K线 + _currentCandle = tick; + } + + notifyListeners(); + } + + /// WebSocket 断连时降级为 HTTP 轮询 + void _startPolling() { + _isConnected = false; + notifyListeners(); + + _pollingTimer?.cancel(); + _pollingTimer = Timer.periodic( + const Duration(seconds: 5), + (_) => _pollCurrentCandle(), + ); + } + + Future _pollCurrentCandle() async { + try { + final response = await _klineService.fetchCurrentCandle( + coinCode: _coinCode, + interval: _interval, + ); + if (response.success && response.data != null) { + _currentCandle = response.data; + notifyListeners(); + } + } catch (_) {} + } + + @override + void dispose() { + _wsSubscription?.cancel(); + _wsService.unsubscribe(_coinCode); + _pollingTimer?.cancel(); + super.dispose(); + } +} diff --git a/flutter_monisuo/lib/ui/pages/kline/components/interval_selector.dart b/flutter_monisuo/lib/ui/pages/kline/components/interval_selector.dart new file mode 100644 index 0000000..cd606e2 --- /dev/null +++ b/flutter_monisuo/lib/ui/pages/kline/components/interval_selector.dart @@ -0,0 +1,56 @@ +import 'package:flutter/material.dart'; +import '../../../../core/theme/app_theme.dart'; +import '../../../../core/theme/app_theme_extension.dart'; + +/// K线周期选择器:15m / 1h / 4h / 1d / 1M +class IntervalSelector extends StatelessWidget { + final String selected; + final ValueChanged onChanged; + + static const List> intervals = [ + MapEntry('15m', '15分'), + MapEntry('1h', '1时'), + MapEntry('4h', '4时'), + MapEntry('1d', '日线'), + MapEntry('1M', '月线'), + ]; + + const IntervalSelector({ + super.key, + required this.selected, + required this.onChanged, + }); + + @override + Widget build(BuildContext context) { + return Row( + children: intervals.map((e) { + final isSelected = e.key == selected; + return Expanded( + child: GestureDetector( + onTap: () => onChanged(e.key), + child: Container( + padding: const EdgeInsets.symmetric(vertical: 8), + decoration: BoxDecoration( + color: isSelected + ? context.colors.primary.withValues(alpha: 0.15) + : Colors.transparent, + borderRadius: BorderRadius.circular(6), + ), + alignment: Alignment.center, + child: Text( + e.value, + style: AppTextStyles.bodyMedium(context).copyWith( + color: isSelected + ? context.colors.primary + : context.appColors.onSurfaceMuted, + fontWeight: isSelected ? FontWeight.w600 : FontWeight.w400, + ), + ), + ), + ), + ); + }).toList(), + ); + } +} diff --git a/flutter_monisuo/lib/ui/pages/kline/components/kline_stats_bar.dart b/flutter_monisuo/lib/ui/pages/kline/components/kline_stats_bar.dart new file mode 100644 index 0000000..639ec1a --- /dev/null +++ b/flutter_monisuo/lib/ui/pages/kline/components/kline_stats_bar.dart @@ -0,0 +1,77 @@ +import 'package:flutter/material.dart'; +import '../../../../core/theme/app_theme.dart'; +import '../../../../core/theme/app_theme_extension.dart'; +import '../../../../data/models/kline_candle.dart'; + +/// K线 OHLC 信息栏 +class KlineStatsBar extends StatelessWidget { + final KlineCandle? candle; + + const KlineStatsBar({super.key, this.candle}); + + @override + Widget build(BuildContext context) { + if (candle == null) return const SizedBox.shrink(); + + final c = candle!; + final change = c.closePrice - c.openPrice; + final changePct = c.openPrice > 0 ? (change / c.openPrice * 100) : 0.0; + final isUp = change >= 0; + final color = isUp ? context.appColors.up : context.appColors.down; + + return Container( + padding: const EdgeInsets.symmetric(horizontal: 12, vertical: 6), + child: Row( + children: [ + _statItem(context, '开', _fmt(c.openPrice), color), + const SizedBox(width: 12), + _statItem(context, '高', _fmt(c.highPrice), color), + const SizedBox(width: 12), + _statItem(context, '低', _fmt(c.lowPrice), color), + const SizedBox(width: 12), + _statItem(context, '收', _fmt(c.closePrice), color), + const SizedBox(width: 12), + _statItem(context, '量', _fmtVol(c.volume), color), + const Spacer(), + Text( + '${isUp ? '+' : ''}${changePct.toStringAsFixed(2)}%', + style: AppTextStyles.labelLarge(context).copyWith( + color: color, + fontWeight: FontWeight.bold, + ), + ), + ], + ), + ); + } + + Widget _statItem(BuildContext context, String label, String value, Color color) { + return Row( + mainAxisSize: MainAxisSize.min, + children: [ + Text(label, + style: AppTextStyles.bodySmall(context).copyWith( + color: context.appColors.onSurfaceMuted, + )), + const SizedBox(width: 2), + Text(value, + style: AppTextStyles.bodySmall(context).copyWith( + color: color, + fontWeight: FontWeight.w500, + )), + ], + ); + } + + String _fmt(double v) { + if (v >= 1000) return v.toStringAsFixed(2); + if (v >= 1) return v.toStringAsFixed(4); + return v.toStringAsFixed(6); + } + + String _fmtVol(double v) { + if (v >= 1000000) return '${(v / 1000000).toStringAsFixed(1)}M'; + if (v >= 1000) return '${(v / 1000).toStringAsFixed(1)}K'; + return v.toStringAsFixed(0); + } +} diff --git a/flutter_monisuo/lib/ui/pages/kline/kline_page.dart b/flutter_monisuo/lib/ui/pages/kline/kline_page.dart new file mode 100644 index 0000000..0ee245c --- /dev/null +++ b/flutter_monisuo/lib/ui/pages/kline/kline_page.dart @@ -0,0 +1,280 @@ +import 'package:flutter/material.dart'; +import 'package:provider/provider.dart'; +import 'package:shadcn_ui/shadcn_ui.dart'; +import 'package:k_chart/flutter_k_chart.dart'; +import '../../../core/theme/app_theme.dart'; +import '../../../core/theme/app_theme_extension.dart'; +import '../../../core/theme/app_spacing.dart'; +import '../../../data/models/coin.dart'; +import '../../../providers/kline_provider.dart'; +import '../main/main_page.dart'; +import 'components/interval_selector.dart'; +import 'components/kline_stats_bar.dart'; + +/// K线图表页面 +class KlinePage extends StatefulWidget { + final Coin coin; + + const KlinePage({super.key, required this.coin}); + + @override + State createState() => _KlinePageState(); +} + +class _KlinePageState extends State { + List? _kLineEntities; + final ChartColors _chartColors = ChartColors(); + final ChartStyle _chartStyle = ChartStyle(); + + @override + void initState() { + super.initState(); + } + + @override + Widget build(BuildContext context) { + final isDark = context.isDark; + _chartColors.bgColor = [isDark ? const Color(0xff1a1a2e) : Colors.white, isDark ? const Color(0xff1a1a2e) : Colors.white]; + _chartColors.gridColor = isDark ? const Color(0xff2d2d44) : const Color(0xffe0e0e0); + _chartColors.upColor = context.appColors.up; + _chartColors.dnColor = context.appColors.down; + + return Scaffold( + backgroundColor: context.colors.background, + appBar: AppBar( + backgroundColor: context.colors.surface, + elevation: 0, + scrolledUnderElevation: 0, + leading: IconButton( + icon: Icon(LucideIcons.arrowLeft, color: context.colors.onSurface), + onPressed: () => Navigator.of(context).pop(), + ), + title: Row( + children: [ + Text(widget.coin.code, + style: AppTextStyles.headlineLarge(context).copyWith( + fontWeight: FontWeight.bold, + )), + const SizedBox(width: 8), + Text(widget.coin.formattedPrice, + style: AppTextStyles.headlineMedium(context).copyWith( + color: context.colors.primary, + )), + const SizedBox(width: 6), + _ChangeBadge(coin: widget.coin), + ], + ), + ), + body: Consumer( + builder: (context, provider, _) { + // 数据转换 + _updateEntities(provider); + + return Column( + children: [ + // 周期选择器 + Padding( + padding: const EdgeInsets.symmetric( + horizontal: AppSpacing.md, + vertical: AppSpacing.sm, + ), + child: IntervalSelector( + selected: provider.interval, + onChanged: (v) => provider.changeInterval(v), + ), + ), + // OHLC 信息栏 + KlineStatsBar(candle: provider.currentCandle), + const Divider(height: 1), + // K线图表 + Expanded( + child: _buildChart(provider), + ), + // 底部操作栏 + _BottomActionBar(coin: widget.coin), + ], + ); + }, + ), + ); + } + + void _updateEntities(KlineProvider provider) { + final allCandles = [...provider.candles]; + if (provider.currentCandle != null) { + allCandles.add(provider.currentCandle!); + } + + if (allCandles.isEmpty) { + _kLineEntities = null; + return; + } + + _kLineEntities = allCandles.map((c) { + return KLineEntity.fromJson({ + 'open': c.openPrice, + 'high': c.highPrice, + 'low': c.lowPrice, + 'close': c.closePrice, + 'vol': c.volume, + 'amount': c.closePrice * c.volume, + 'time': c.openTime, + 'id': c.openTime, + }); + }).toList(); + + DataUtil.calculate(_kLineEntities!); + } + + Widget _buildChart(KlineProvider provider) { + if (provider.isLoading) { + return const Center(child: CircularProgressIndicator()); + } + + if (_kLineEntities == null || _kLineEntities!.isEmpty) { + return Center( + child: Column( + mainAxisAlignment: MainAxisAlignment.center, + children: [ + Icon(LucideIcons.chartNoAxesColumn, + size: 48, + color: context.appColors.onSurfaceMuted.withValues(alpha: 0.4)), + const SizedBox(height: AppSpacing.md), + Text('暂无K线数据', + style: AppTextStyles.headlineMedium(context).copyWith( + color: context.appColors.onSurfaceMuted, + )), + const SizedBox(height: AppSpacing.sm), + ], + ), + ); + } + + return Stack( + children: [ + SizedBox( + height: double.infinity, + width: double.infinity, + child: KChartWidget( + _kLineEntities, + _chartStyle, + _chartColors, + isLine: false, + isTrendLine: false, + mainState: MainState.MA, + volHidden: false, + secondaryState: SecondaryState.MACD, + fixedLength: 4, + timeFormat: TimeFormat.YEAR_MONTH_DAY, + showNowPrice: true, + hideGrid: false, + isTapShowInfoDialog: false, + onSecondaryTap: () {}, + ), + ), + ], + ); + } +} + +/// 涨跌标签 +class _ChangeBadge extends StatelessWidget { + final Coin coin; + const _ChangeBadge({required this.coin}); + + @override + Widget build(BuildContext context) { + final isUp = coin.isUp; + final color = isUp ? context.appColors.up : context.appColors.down; + return Container( + padding: const EdgeInsets.symmetric(horizontal: 6, vertical: 2), + decoration: BoxDecoration( + color: color.withValues(alpha: 0.1), + borderRadius: BorderRadius.circular(4), + ), + child: Text( + coin.formattedChange, + style: AppTextStyles.bodySmall(context).copyWith( + color: color, + fontWeight: FontWeight.w600, + ), + ), + ); + } +} + +/// 底部交易操作栏 +class _BottomActionBar extends StatelessWidget { + final Coin coin; + const _BottomActionBar({required this.coin}); + + @override + Widget build(BuildContext context) { + return Container( + padding: const EdgeInsets.fromLTRB( + AppSpacing.lg, AppSpacing.sm, AppSpacing.lg, AppSpacing.lg, + ), + decoration: BoxDecoration( + color: context.colors.surface, + border: Border( + top: BorderSide( + color: context.colors.outlineVariant.withValues(alpha: 0.2), + ), + ), + ), + child: Row( + children: [ + Expanded( + child: SizedBox( + height: 44, + child: ElevatedButton( + onPressed: () => _navigateToTrade(context, isBuy: true), + style: ElevatedButton.styleFrom( + backgroundColor: context.appColors.up, + foregroundColor: Colors.white, + elevation: 0, + shape: RoundedRectangleBorder( + borderRadius: BorderRadius.circular(AppRadius.lg), + ), + ), + child: Text('买入', + style: AppTextStyles.headlineMedium(context).copyWith( + color: Colors.white, + fontWeight: FontWeight.bold, + )), + ), + ), + ), + const SizedBox(width: AppSpacing.md), + Expanded( + child: SizedBox( + height: 44, + child: ElevatedButton( + onPressed: () => _navigateToTrade(context, isBuy: false), + style: ElevatedButton.styleFrom( + backgroundColor: context.appColors.down, + foregroundColor: Colors.white, + elevation: 0, + shape: RoundedRectangleBorder( + borderRadius: BorderRadius.circular(AppRadius.lg), + ), + ), + child: Text('卖出', + style: AppTextStyles.headlineMedium(context).copyWith( + color: Colors.white, + fontWeight: FontWeight.bold, + )), + ), + ), + ), + ], + ), + ); + } + + void _navigateToTrade(BuildContext context, {required bool isBuy}) { + Navigator.of(context).pop(); + final mainState = context.findAncestorStateOfType(); + mainState?.switchToTrade(coin.code); + } +} diff --git a/monisuo-admin/src/pages/monisuo/kline-config.vue b/monisuo-admin/src/pages/monisuo/kline-config.vue new file mode 100644 index 0000000..c3e30d8 --- /dev/null +++ b/monisuo-admin/src/pages/monisuo/kline-config.vue @@ -0,0 +1,324 @@ + + + diff --git a/monisuo-admin/src/services/api/monisuo-kline.api.ts b/monisuo-admin/src/services/api/monisuo-kline.api.ts new file mode 100644 index 0000000..0983b18 --- /dev/null +++ b/monisuo-admin/src/services/api/monisuo-kline.api.ts @@ -0,0 +1,71 @@ +import { useAxios } from '../../composables/use-axios' + +// ==================== Types ==================== + +export interface KlineConfig { + coinId: number + coinCode: string + coinName: string + simulationEnabled: number // 0 or 1 + tradeStartTime: string // HH:mm + tradeEndTime: string // HH:mm + priceMin: number + priceMax: number + currentPrice: number + priceType: number +} + +export interface KlineConfigUpdate { + coinCode: string + tradeStartTime?: string + tradeEndTime?: string + priceMin?: number + priceMax?: number + simulationEnabled?: number +} + +export interface KlineCandle { + coinCode: string + interval: string + openTime: number + openPrice: number + highPrice: number + lowPrice: number + closePrice: number + volume: number + closeTime: number +} + +// ==================== API Functions ==================== + +/** 获取所有币种K线配置 */ +export async function getKlineConfigs(): Promise { + const { axiosInstance } = useAxios() + const { data } = await axiosInstance.get('/admin/kline/config') + const list: any[] = (data as any)?.data ?? [] + return list.map(item => ({ + coinId: item.id, + coinCode: item.code, + coinName: item.name, + simulationEnabled: item.simulationEnabled ?? 0, + tradeStartTime: item.tradeStartTime, + tradeEndTime: item.tradeEndTime, + priceMin: item.priceMin, + priceMax: item.priceMax, + currentPrice: item.price, + priceType: item.priceType, + })) +} + +/** 保存K线配置 */ +export async function saveKlineConfig(config: KlineConfigUpdate): Promise { + const { axiosInstance } = useAxios() + await axiosInstance.post('/admin/kline/config', config) +} + +/** 获取K线预览数据(用于 echarts) */ +export async function getKlinePreview(coinCode: string, interval: string = '1h', limit: number = 100): Promise { + const { axiosInstance } = useAxios() + const { data } = await axiosInstance.get('/admin/kline/preview', { params: { coinCode, interval, limit } }) + return (data as any)?.data ?? [] +} diff --git a/sql/patch_coin_kline.sql b/sql/patch_coin_kline.sql new file mode 100644 index 0000000..0d2886b --- /dev/null +++ b/sql/patch_coin_kline.sql @@ -0,0 +1,28 @@ +-- ============================================= +-- K线模拟功能 — 数据库补丁 +-- ============================================= + +-- 1. 新建 coin_kline 表(K线蜡烛数据) +CREATE TABLE IF NOT EXISTS `coin_kline` ( + `id` bigint(20) NOT NULL AUTO_INCREMENT, + `coin_code` varchar(20) NOT NULL COMMENT '币种代码', + `interval` varchar(5) NOT NULL COMMENT '周期: 15m/1h/4h/1d/1M', + `open_time` bigint(20) NOT NULL COMMENT '开盘时间戳(ms)', + `open_price` decimal(20,8) NOT NULL, + `high_price` decimal(20,8) NOT NULL, + `low_price` decimal(20,8) NOT NULL, + `close_price` decimal(20,8) NOT NULL, + `volume` decimal(20,4) DEFAULT 0 COMMENT '模拟成交量', + `close_time` bigint(20) NOT NULL COMMENT '收盘时间戳(ms)', + `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uk_coin_interval_open` (`coin_code`, `interval`, `open_time`), + KEY `idx_coin_interval_close` (`coin_code`, `interval`, `close_time`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='K线蜡烛数据'; + +-- 2. coin 表新增K线配置字段 +ALTER TABLE `coin` + ADD COLUMN `trade_start_time` varchar(5) DEFAULT '09:00' COMMENT '交易开始时间 HH:mm', + ADD COLUMN `trade_end_time` varchar(5) DEFAULT '23:00' COMMENT '交易结束时间 HH:mm', + ADD COLUMN `max_change_percent` decimal(5,2) DEFAULT 5.00 COMMENT '每日最大涨跌幅(%)', + ADD COLUMN `simulation_enabled` tinyint(1) DEFAULT 0 COMMENT '1=启用K线模拟'; diff --git a/src/main/java/com/it/rattan/monisuo/config/RedisConfig.java b/src/main/java/com/it/rattan/monisuo/config/RedisConfig.java new file mode 100644 index 0000000..4676881 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/config/RedisConfig.java @@ -0,0 +1,42 @@ +package com.it.rattan.monisuo.config; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.PropertyAccessor; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; + +/** + * Redis 配置 + */ +@Configuration +public class RedisConfig { + + @Bean + public RedisTemplate redisTemplate(RedisConnectionFactory factory) { + RedisTemplate template = new RedisTemplate<>(); + template.setConnectionFactory(factory); + + ObjectMapper om = new ObjectMapper(); + om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); + om.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL); + + GenericJackson2JsonRedisSerializer jsonSerializer = new GenericJackson2JsonRedisSerializer(om); + StringRedisSerializer stringSerializer = new StringRedisSerializer(); + + // key 用 String + template.setKeySerializer(stringSerializer); + template.setHashKeySerializer(stringSerializer); + // value 用 JSON + template.setValueSerializer(jsonSerializer); + template.setHashValueSerializer(jsonSerializer); + + template.afterPropertiesSet(); + return template; + } +} diff --git a/src/main/java/com/it/rattan/monisuo/config/WebSocketConfig.java b/src/main/java/com/it/rattan/monisuo/config/WebSocketConfig.java new file mode 100644 index 0000000..fb2c1e3 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/config/WebSocketConfig.java @@ -0,0 +1,39 @@ +package com.it.rattan.monisuo.config; + +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.simp.config.MessageBrokerRegistry; +import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; +import org.springframework.web.socket.config.annotation.StompEndpointRegistry; +import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; + +/** + * WebSocket 配置 — K线实时推送 + * + * 端点: /ws/kline + * 主题: /topic/kline/{coinCode} + * + * Flutter 使用原生 STOMP 连接,浏览器可用 SockJS 降级 + */ +@Configuration +@EnableWebSocketMessageBroker +public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { + + @Override + public void configureMessageBroker(MessageBrokerRegistry config) { + // 服务端推送主题前缀 + config.enableSimpleBroker("/topic"); + // 客户端发送前缀(本场景不需要客户端主动发消息,预留) + config.setApplicationDestinationPrefixes("/app"); + } + + @Override + public void registerStompEndpoints(StompEndpointRegistry registry) { + // 原生 WebSocket 端点(Flutter 用) + registry.addEndpoint("/ws/kline") + .setAllowedOrigins("*"); + // SockJS 降级端点(浏览器用) + registry.addEndpoint("/ws/kline") + .setAllowedOrigins("*") + .withSockJS(); + } +} diff --git a/src/main/java/com/it/rattan/monisuo/controller/AdminKlineController.java b/src/main/java/com/it/rattan/monisuo/controller/AdminKlineController.java new file mode 100644 index 0000000..efcb642 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/controller/AdminKlineController.java @@ -0,0 +1,66 @@ +package com.it.rattan.monisuo.controller; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.it.rattan.monisuo.common.Result; +import com.it.rattan.monisuo.dto.KlineConfigUpdate; +import com.it.rattan.monisuo.entity.Coin; +import com.it.rattan.monisuo.entity.CoinKline; +import com.it.rattan.monisuo.mapper.CoinMapper; +import com.it.rattan.monisuo.service.CoinKlineService; +import com.it.rattan.monisuo.service.CoinService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.*; + +import java.util.List; + +/** + * 管理端K线配置 API + */ +@Slf4j +@RestController +public class AdminKlineController { + + @Autowired + private CoinKlineService coinKlineService; + + @Autowired + private CoinMapper coinMapper; + + /** + * 获取所有币种K线配置 + */ + @GetMapping("/admin/kline/config") + public Result> getKlineConfigs() { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.orderByDesc(Coin::getSort); + List coins = coinMapper.selectList(wrapper); + return Result.success(coins); + } + + /** + * 保存币种K线配置 + */ + @PostMapping("/admin/kline/config") + public Result saveKlineConfig(@RequestBody KlineConfigUpdate config) { + try { + coinKlineService.updateConfig(config); + return Result.success("配置已保存", null); + } catch (Exception e) { + return Result.fail(e.getMessage()); + } + } + + /** + * 获取K线预览数据(用于 admin echarts 图表) + */ + @GetMapping("/admin/kline/preview") + public Result> getKlinePreview( + @RequestParam String coinCode, + @RequestParam(defaultValue = "1h") String interval, + @RequestParam(defaultValue = "100") int limit) { + List candles = coinKlineService.getHistory( + coinCode.toUpperCase(), interval, limit, null); + return Result.success(candles); + } +} diff --git a/src/main/java/com/it/rattan/monisuo/controller/KlineController.java b/src/main/java/com/it/rattan/monisuo/controller/KlineController.java new file mode 100644 index 0000000..bf34256 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/controller/KlineController.java @@ -0,0 +1,69 @@ +package com.it.rattan.monisuo.controller; + +import com.it.rattan.monisuo.common.Result; +import com.it.rattan.monisuo.entity.CoinKline; +import com.it.rattan.monisuo.service.CoinKlineService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.*; + +import java.util.List; +import java.util.Map; +import java.util.HashMap; + +/** + * 用户端K线 API + */ +@RestController +public class KlineController { + + @Autowired + private CoinKlineService coinKlineService; + + /** + * 获取支持的K线周期列表 + */ + @GetMapping("/api/kline/intervals") + public Result> getIntervals() { + return Result.success(coinKlineService.getIntervals()); + } + + /** + * 获取历史K线数据(分页) + * + * @param coinCode 币种代码 + * @param interval 周期 (15m/1h/4h/1d/1M) + * @param limit 数量 (默认200, 最大500) + * @param before 查询此时间戳(ms)之前的K线(分页用) + */ + @GetMapping("/api/kline/history") + public Result> getHistory( + @RequestParam String coinCode, + @RequestParam(defaultValue = "1h") String interval, + @RequestParam(defaultValue = "200") int limit, + @RequestParam(required = false) Long before) { + + List candles = coinKlineService.getHistory( + coinCode.toUpperCase(), interval, limit, before); + + Map data = new HashMap<>(); + data.put("list", candles); + data.put("coinCode", coinCode.toUpperCase()); + data.put("interval", interval); + return Result.success(data); + } + + /** + * 获取当前进行中的K线 + */ + @GetMapping("/api/kline/current") + public Result getCurrentCandle( + @RequestParam String coinCode, + @RequestParam(defaultValue = "1h") String interval) { + CoinKline candle = coinKlineService.getCurrentCandle( + coinCode.toUpperCase(), interval); + if (candle == null) { + return Result.fail("暂无K线数据"); + } + return Result.success(candle); + } +} diff --git a/src/main/java/com/it/rattan/monisuo/dto/KlineConfigUpdate.java b/src/main/java/com/it/rattan/monisuo/dto/KlineConfigUpdate.java new file mode 100644 index 0000000..4fbb457 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/dto/KlineConfigUpdate.java @@ -0,0 +1,37 @@ +package com.it.rattan.monisuo.dto; + +import lombok.Getter; +import lombok.Setter; +import java.io.Serializable; +import java.math.BigDecimal; + +/** + * 管理员K线配置更新 DTO + */ +@Getter +@Setter +public class KlineConfigUpdate implements Serializable { + + private static final long serialVersionUID = 1L; + + /** 币种代码 */ + private String coinCode; + + /** 交易开始时间 HH:mm */ + private String tradeStartTime; + + /** 交易结束时间 HH:mm */ + private String tradeEndTime; + + /** 每日最大涨跌幅(%) */ + private BigDecimal maxChangePercent; + + /** K线模拟最低价 */ + private BigDecimal priceMin; + + /** K线模拟最高价 */ + private BigDecimal priceMax; + + /** 1=启用模拟 */ + private Integer simulationEnabled; +} diff --git a/src/main/java/com/it/rattan/monisuo/dto/KlineTick.java b/src/main/java/com/it/rattan/monisuo/dto/KlineTick.java new file mode 100644 index 0000000..b224a77 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/dto/KlineTick.java @@ -0,0 +1,67 @@ +package com.it.rattan.monisuo.dto; + +import lombok.Getter; +import lombok.Setter; +import java.io.Serializable; +import java.math.BigDecimal; + +/** + * K线 Tick 推送 DTO(WebSocket 广播用) + */ +@Getter +@Setter +public class KlineTick implements Serializable { + + private static final long serialVersionUID = 1L; + + /** 币种代码 */ + private String coinCode; + + /** 周期 */ + private String interval; + + /** 开盘时间戳(ms) */ + private Long openTime; + + /** 开盘价 */ + private BigDecimal openPrice; + + /** 最高价 */ + private BigDecimal highPrice; + + /** 最低价 */ + private BigDecimal lowPrice; + + /** 收盘价(最新价) */ + private BigDecimal closePrice; + + /** 成交量 */ + private BigDecimal volume; + + /** 收盘时间戳(ms) */ + private Long closeTime; + + /** 是否已收盘 */ + private Boolean isClosed; + + /** 当前时间戳 */ + private Long timestamp; + + public KlineTick() {} + + public static KlineTick fromEntity(com.it.rattan.monisuo.entity.CoinKline kline, boolean isClosed) { + KlineTick tick = new KlineTick(); + tick.setCoinCode(kline.getCoinCode()); + tick.setInterval(kline.getInterval()); + tick.setOpenTime(kline.getOpenTime()); + tick.setOpenPrice(kline.getOpenPrice()); + tick.setHighPrice(kline.getHighPrice()); + tick.setLowPrice(kline.getLowPrice()); + tick.setClosePrice(kline.getClosePrice()); + tick.setVolume(kline.getVolume()); + tick.setCloseTime(kline.getCloseTime()); + tick.setIsClosed(isClosed); + tick.setTimestamp(System.currentTimeMillis()); + return tick; + } +} diff --git a/src/main/java/com/it/rattan/monisuo/entity/CoinKline.java b/src/main/java/com/it/rattan/monisuo/entity/CoinKline.java new file mode 100644 index 0000000..123af04 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/entity/CoinKline.java @@ -0,0 +1,53 @@ +package com.it.rattan.monisuo.entity; + +import com.baomidou.mybatisplus.annotation.*; +import lombok.Getter; +import lombok.Setter; +import java.io.Serializable; +import java.math.BigDecimal; +import java.time.LocalDateTime; + +/** + * K线蜡烛数据实体 + */ +@Getter +@Setter +@TableName("coin_kline") +public class CoinKline implements Serializable { + + private static final long serialVersionUID = 1L; + + @TableId(type = IdType.AUTO) + private Long id; + + /** 币种代码 */ + private String coinCode; + + /** 周期: 15m/1h/4h/1d/1M */ + private String interval; + + /** 开盘时间戳(ms) */ + private Long openTime; + + /** 开盘价 */ + private BigDecimal openPrice; + + /** 最高价 */ + private BigDecimal highPrice; + + /** 最低价 */ + private BigDecimal lowPrice; + + /** 收盘价 */ + private BigDecimal closePrice; + + /** 模拟成交量 */ + private BigDecimal volume; + + /** 收盘时间戳(ms) */ + private Long closeTime; + + /** 创建时间 */ + @TableField(fill = FieldFill.INSERT) + private LocalDateTime createTime; +} diff --git a/src/main/java/com/it/rattan/monisuo/mapper/CoinKlineMapper.java b/src/main/java/com/it/rattan/monisuo/mapper/CoinKlineMapper.java new file mode 100644 index 0000000..dbe885a --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/mapper/CoinKlineMapper.java @@ -0,0 +1,37 @@ +package com.it.rattan.monisuo.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.it.rattan.monisuo.entity.CoinKline; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; +import java.util.List; + +/** + * K线数据 Mapper + */ +@Mapper +public interface CoinKlineMapper extends BaseMapper { + + /** + * 查询最近的K线数据(按时间倒序) + */ + @Select("SELECT * FROM coin_kline WHERE coin_code = #{coinCode} " + + "AND `interval` = #{interval} " + + "AND close_time < #{beforeCloseTime} " + + "ORDER BY open_time DESC LIMIT #{limit}") + List selectRecentCandles(@Param("coinCode") String coinCode, + @Param("interval") String interval, + @Param("beforeCloseTime") long beforeCloseTime, + @Param("limit") int limit); + + /** + * 查询最近的K线数据(不限 before) + */ + @Select("SELECT * FROM coin_kline WHERE coin_code = #{coinCode} " + + "AND `interval` = #{interval} " + + "ORDER BY open_time DESC LIMIT #{limit}") + List selectLatestCandles(@Param("coinCode") String coinCode, + @Param("interval") String interval, + @Param("limit") int limit); +} diff --git a/src/main/java/com/it/rattan/monisuo/service/CoinKlineService.java b/src/main/java/com/it/rattan/monisuo/service/CoinKlineService.java new file mode 100644 index 0000000..756bde4 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/service/CoinKlineService.java @@ -0,0 +1,482 @@ +package com.it.rattan.monisuo.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.it.rattan.monisuo.dto.KlineConfigUpdate; +import com.it.rattan.monisuo.dto.KlineTick; +import com.it.rattan.monisuo.entity.Coin; +import com.it.rattan.monisuo.entity.CoinKline; +import com.it.rattan.monisuo.mapper.CoinKlineMapper; +import com.it.rattan.monisuo.mapper.CoinMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.messaging.simp.SimpMessagingTemplate; +import org.springframework.stereotype.Service; + +import javax.annotation.PostConstruct; +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.time.*; +import java.time.format.DateTimeFormatter; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ThreadLocalRandom; + +/** + * K线模拟引擎 + * + * 双写策略:ConcurrentHashMap(高性能读写) + Redis(持久化/重启恢复) + */ +@Slf4j +@Service +public class CoinKlineService { + + private static final String REDIS_CANDLE_PREFIX = "kline:candle:"; + private static final String REDIS_PRICE_PREFIX = "kline:price:"; + + /** 本地内存 — tick 级别高频读写 */ + private final ConcurrentHashMap currentCandles = new ConcurrentHashMap<>(); + + private static final List INTERVALS = Arrays.asList("15m", "1h", "4h", "1d", "1M"); + private static final BigDecimal TICK_VOLUME_BASE = new BigDecimal("100"); + private static final double DAILY_TICKS = 9600.0; + + @Autowired + private CoinMapper coinMapper; + + @Autowired + private CoinKlineMapper coinKlineMapper; + + @Autowired + private CoinService coinService; + + @Autowired(required = false) + private SimpMessagingTemplate messagingTemplate; + + @Autowired + private RedisTemplate redisTemplate; + + /** + * 启动时从 Redis 恢复当前进行中的蜡烛 + */ + @PostConstruct + public void init() { + Set keys = redisTemplate.keys(REDIS_CANDLE_PREFIX + "*"); + if (keys != null && !keys.isEmpty()) { + for (String key : keys) { + try { + Object obj = redisTemplate.opsForValue().get(key); + if (obj instanceof CoinKline) { + CoinKline candle = (CoinKline) obj; + String mapKey = candle.getCoinCode() + ":" + candle.getInterval(); + currentCandles.put(mapKey, candle); + } + } catch (Exception e) { + log.warn("恢复K线数据失败: key={}, error={}", key, e.getMessage()); + } + } + log.info("从Redis恢复了 {} 个进行中的K线蜡烛", currentCandles.size()); + } + + // 恢复模拟价格到 CoinService 缓存 + Set priceKeys = redisTemplate.keys(REDIS_PRICE_PREFIX + "*"); + if (priceKeys != null && !priceKeys.isEmpty()) { + for (String key : priceKeys) { + try { + String coinCode = key.substring(REDIS_PRICE_PREFIX.length()); + Object priceObj = redisTemplate.opsForValue().get(key); + if (priceObj instanceof BigDecimal) { + coinService.updateCachedPrice(coinCode, (BigDecimal) priceObj); + } + } catch (Exception e) { + log.warn("恢复价格失败: key={}, error={}", key, e.getMessage()); + } + } + log.info("从Redis恢复了 {} 个币种的模拟价格", priceKeys.size()); + } + } + + // ============================ 公开方法 ============================ + + public List getIntervals() { + return INTERVALS; + } + + /** + * 获取历史K线(分页) + */ + public List getHistory(String coinCode, String interval, int limit, Long beforeTime) { + if (limit <= 0 || limit > 500) limit = 200; + if (beforeTime == null || beforeTime <= 0) { + beforeTime = System.currentTimeMillis(); + } + List candles = coinKlineMapper.selectRecentCandles(coinCode, interval, beforeTime, limit); + Collections.reverse(candles); + return candles; + } + + /** + * 获取当前进行中的K线 + */ + public CoinKline getCurrentCandle(String coinCode, String interval) { + String key = coinCode.toUpperCase() + ":" + interval; + CoinKline kline = currentCandles.get(key); + if (kline != null) { + return copyKline(kline); + } + List latest = coinKlineMapper.selectLatestCandles(coinCode.toUpperCase(), interval, 1); + if (!latest.isEmpty()) { + return latest.get(0); + } + return null; + } + + /** + * 获取所有启用模拟的币种 + */ + public List getSimulatedCoins() { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.eq(Coin::getSimulationEnabled, 1) + .eq(Coin::getStatus, 1); + return coinMapper.selectList(wrapper); + } + + /** + * 管理员更新K线配置 + */ + public void updateConfig(KlineConfigUpdate config) { + Coin coin = coinService.getCoinByCode(config.getCoinCode()); + if (coin == null) throw new RuntimeException("币种不存在"); + + if (config.getTradeStartTime() != null) coin.setTradeStartTime(config.getTradeStartTime()); + if (config.getTradeEndTime() != null) coin.setTradeEndTime(config.getTradeEndTime()); + if (config.getPriceMin() != null) coin.setPriceMin(config.getPriceMin()); + if (config.getPriceMax() != null) coin.setPriceMax(config.getPriceMax()); + if (config.getSimulationEnabled() != null) coin.setSimulationEnabled(config.getSimulationEnabled()); + + if (coin.getPriceMin() != null && coin.getPriceMax() != null + && coin.getPriceMin().compareTo(coin.getPriceMax()) >= 0) { + throw new RuntimeException("最低价必须小于最高价"); + } + + coin.setUpdateTime(LocalDateTime.now()); + coinMapper.updateById(coin); + coinService.clearCache(coin.getCode()); + + log.info("K线配置已更新: {} simulation={} priceMin={} priceMax={}", + coin.getCode(), coin.getSimulationEnabled(), coin.getPriceMin(), coin.getPriceMax()); + } + + // ============================ Tick 生成 ============================ + + /** + * 每 3 秒调用:为所有启用模拟的币种生成一个价格 tick + */ + public List generateTicks() { + List coins = getSimulatedCoins(); + if (coins.isEmpty()) return Collections.emptyList(); + + LocalTime now = LocalTime.now(); + List ticks = new ArrayList<>(); + + for (Coin coin : coins) { + if (!isInTradingHours(coin, now)) continue; + if (coin.getPriceMin() == null || coin.getPriceMax() == null) continue; + if (coin.getPriceMin().compareTo(coin.getPriceMax()) >= 0) continue; + + BigDecimal currentPrice = coin.getPrice(); + if (currentPrice == null || currentPrice.compareTo(BigDecimal.ZERO) <= 0) { + currentPrice = coin.getPriceMin().add(coin.getPriceMax()) + .divide(new BigDecimal("2"), 8, RoundingMode.DOWN); + } + final BigDecimal tickOpenPrice = currentPrice; + + BigDecimal newPrice = simulatePrice(coin, currentPrice); + + // 更新内存价格 + coin.setPrice(newPrice); + + // 同步到 CoinService 本地缓存 + Redis + coinService.updateCachedPrice(coin.getCode(), newPrice); + savePriceToRedis(coin.getCode(), newPrice); + + // 为每个周期更新内存K线 + for (String interval : INTERVALS) { + String key = coin.getCode() + ":" + interval; + CoinKline candle = currentCandles.computeIfAbsent(key, k -> { + long[] times = calculateIntervalTimes(interval); + CoinKline c = new CoinKline(); + c.setCoinCode(coin.getCode()); + c.setInterval(interval); + c.setOpenTime(times[0]); + c.setCloseTime(times[1]); + c.setOpenPrice(tickOpenPrice); + c.setHighPrice(tickOpenPrice); + c.setLowPrice(tickOpenPrice); + c.setClosePrice(tickOpenPrice); + c.setVolume(BigDecimal.ZERO); + return c; + }); + + candle.setClosePrice(newPrice); + if (newPrice.compareTo(candle.getHighPrice()) > 0) { + candle.setHighPrice(newPrice); + } + if (newPrice.compareTo(candle.getLowPrice()) < 0) { + candle.setLowPrice(newPrice); + } + BigDecimal volIncrement = TICK_VOLUME_BASE.multiply( + new BigDecimal(ThreadLocalRandom.current().nextDouble(0.5, 2.0)) + .setScale(4, RoundingMode.DOWN) + ); + candle.setVolume(candle.getVolume().add(volIncrement)); + + // 每 tick 写入 Redis(异步安全) + saveCandleToRedis(key, candle); + } + + // 为所有周期生成 tick 事件 + for (String interval : INTERVALS) { + String tickKey = coin.getCode() + ":" + interval; + CoinKline tickCandle = currentCandles.get(tickKey); + if (tickCandle != null) { + ticks.add(KlineTick.fromEntity(tickCandle, false)); + } + } + } + + return ticks; + } + + // ============================ K线收盘 ============================ + + public List closeCandlesForInterval(String closedInterval) { + List coins = getSimulatedCoins(); + List closedTicks = new ArrayList<>(); + + for (Coin coin : coins) { + String key = coin.getCode() + ":" + closedInterval; + CoinKline candle = currentCandles.remove(key); + if (candle == null) continue; + + if (candle.getClosePrice() != null && candle.getClosePrice().compareTo(BigDecimal.ZERO) > 0) { + coinKlineMapper.insert(candle); + log.debug("K线已收盘: {} {} open={}", coin.getCode(), closedInterval, candle.getOpenPrice()); + } + + // 从 Redis 移除已收盘的蜡烛 + removeCandleFromRedis(key); + + closedTicks.add(KlineTick.fromEntity(candle, true)); + + if ("1d".equals(closedInterval)) { + updateCoinDailyStats(coin, candle); + } + } + + return closedTicks; + } + + public List detectIntervalCrossings(long previousTickTime) { + List crossed = new ArrayList<>(); + long now = System.currentTimeMillis(); + + if (differentIntervalSlot(previousTickTime, now, 15 * 60 * 1000L)) crossed.add("15m"); + if (differentIntervalSlot(previousTickTime, now, 60 * 60 * 1000L)) crossed.add("1h"); + if (different4hSlot(previousTickTime, now)) crossed.add("4h"); + if (differentDaySlot(previousTickTime, now)) crossed.add("1d"); + if (differentMonthSlot(previousTickTime, now)) crossed.add("1M"); + + return crossed; + } + + // ============================ Redis 操作 ============================ + + private void saveCandleToRedis(String key, CoinKline candle) { + try { + redisTemplate.opsForValue().set(REDIS_CANDLE_PREFIX + key, candle); + } catch (Exception e) { + log.warn("Redis写入蜡烛失败: key={}, error={}", key, e.getMessage()); + } + } + + private void removeCandleFromRedis(String key) { + try { + redisTemplate.delete(REDIS_CANDLE_PREFIX + key); + } catch (Exception e) { + log.warn("Redis删除蜡烛失败: key={}, error={}", key, e.getMessage()); + } + } + + private void savePriceToRedis(String coinCode, BigDecimal price) { + try { + redisTemplate.opsForValue().set(REDIS_PRICE_PREFIX + coinCode.toUpperCase(), price); + } catch (Exception e) { + log.warn("Redis写入价格失败: coin={}, error={}", coinCode, e.getMessage()); + } + } + + // ============================ GBM 模拟算法 ============================ + + private BigDecimal simulatePrice(Coin coin, BigDecimal currentPrice) { + double S = currentPrice.doubleValue(); + double priceMin = coin.getPriceMin().doubleValue(); + double priceMax = coin.getPriceMax().doubleValue(); + double midpoint = (priceMin + priceMax) / 2.0; + double halfRange = (priceMax - priceMin) / 2.0; + + double sigma = halfRange / Math.max(midpoint, 0.00000001) / Math.sqrt(DAILY_TICKS); + double dt = 1.0 / DAILY_TICKS; + + double u1 = ThreadLocalRandom.current().nextDouble(); + double u2 = ThreadLocalRandom.current().nextDouble(); + double Z = Math.sqrt(-2.0 * Math.log(Math.max(u1, 1e-10))) * Math.cos(2.0 * Math.PI * u2); + + double mu = 0; + double deviation = (S - midpoint) / halfRange; + double absDeviation = Math.abs(deviation); + if (absDeviation > 0.85) { + mu = -deviation * 0.02; + } else if (absDeviation > 0.65) { + mu = -deviation * 0.005; + } + + double dS = mu * S * dt + sigma * S * Math.sqrt(dt) * Z; + double newPrice = S + dS; + + newPrice = Math.max(priceMin, Math.min(priceMax, newPrice)); + newPrice = Math.max(newPrice, 0.00000001); + + return new BigDecimal(newPrice).setScale(8, RoundingMode.DOWN); + } + + // ============================ 辅助方法 ============================ + + private boolean isInTradingHours(Coin coin, LocalTime now) { + String startStr = coin.getTradeStartTime(); + String endStr = coin.getTradeEndTime(); + if (startStr == null || endStr == null) return true; + + try { + LocalTime start = LocalTime.parse(startStr, DateTimeFormatter.ofPattern("HH:mm")); + LocalTime end = LocalTime.parse(endStr, DateTimeFormatter.ofPattern("HH:mm")); + if (end.isAfter(start)) { + return !now.isBefore(start) && !now.isAfter(end); + } else { + return !now.isBefore(start) || !now.isAfter(end); + } + } catch (Exception e) { + return true; + } + } + + private long[] calculateIntervalTimes(String interval) { + Instant now = Instant.now(); + ZoneId zone = ZoneId.of("Asia/Shanghai"); + ZonedDateTime zdt = now.atZone(zone); + + long openTime, closeTime; + switch (interval) { + case "15m": { + int minute = zdt.getMinute(); + int slot = (minute / 15) * 15; + ZonedDateTime open = zdt.withMinute(slot).withSecond(0).withNano(0); + openTime = open.toInstant().toEpochMilli(); + closeTime = open.plusMinutes(15).toInstant().toEpochMilli(); + break; + } + case "1h": { + ZonedDateTime open = zdt.withMinute(0).withSecond(0).withNano(0); + openTime = open.toInstant().toEpochMilli(); + closeTime = open.plusHours(1).toInstant().toEpochMilli(); + break; + } + case "4h": { + int hour = zdt.getHour(); + int slot = (hour / 4) * 4; + ZonedDateTime open = zdt.withHour(slot).withMinute(0).withSecond(0).withNano(0); + openTime = open.toInstant().toEpochMilli(); + closeTime = open.plusHours(4).toInstant().toEpochMilli(); + break; + } + case "1d": { + ZonedDateTime open = zdt.withHour(0).withMinute(0).withSecond(0).withNano(0); + openTime = open.toInstant().toEpochMilli(); + closeTime = open.plusDays(1).toInstant().toEpochMilli(); + break; + } + case "1M": { + ZonedDateTime open = zdt.withDayOfMonth(1).withHour(0).withMinute(0).withSecond(0).withNano(0); + openTime = open.toInstant().toEpochMilli(); + closeTime = open.plusMonths(1).toInstant().toEpochMilli(); + break; + } + default: + throw new IllegalArgumentException("不支持的周期: " + interval); + } + return new long[]{openTime, closeTime}; + } + + private boolean differentIntervalSlot(long t1, long t2, long intervalMs) { + return (t1 / intervalMs) != (t2 / intervalMs); + } + + private boolean different4hSlot(long t1, long t2) { + ZoneId zone = ZoneId.of("Asia/Shanghai"); + int h1 = Instant.ofEpochMilli(t1).atZone(zone).getHour() / 4; + int h2 = Instant.ofEpochMilli(t2).atZone(zone).getHour() / 4; + int d1 = Instant.ofEpochMilli(t1).atZone(zone).getDayOfYear(); + int d2 = Instant.ofEpochMilli(t2).atZone(zone).getDayOfYear(); + return d1 != d2 || h1 != h2; + } + + private boolean differentDaySlot(long t1, long t2) { + ZoneId zone = ZoneId.of("Asia/Shanghai"); + return Instant.ofEpochMilli(t1).atZone(zone).toLocalDate() + .isBefore(Instant.ofEpochMilli(t2).atZone(zone).toLocalDate()); + } + + private boolean differentMonthSlot(long t1, long t2) { + ZoneId zone = ZoneId.of("Asia/Shanghai"); + ZonedDateTime z1 = Instant.ofEpochMilli(t1).atZone(zone); + ZonedDateTime z2 = Instant.ofEpochMilli(t2).atZone(zone); + return z1.getYear() != z2.getYear() || z1.getMonthValue() != z2.getMonthValue(); + } + + private void updateCoinDailyStats(Coin coin, CoinKline dailyCandle) { + coin.setPrice(dailyCandle.getClosePrice()); + BigDecimal change = dailyCandle.getClosePrice() + .subtract(dailyCandle.getOpenPrice()) + .divide(dailyCandle.getOpenPrice(), 8, RoundingMode.DOWN) + .multiply(new BigDecimal("100")); + coin.setChange24h(change); + coin.setHigh24h(dailyCandle.getHighPrice()); + coin.setLow24h(dailyCandle.getLowPrice()); + coin.setVolume24h(dailyCandle.getVolume()); + coin.setUpdateTime(LocalDateTime.now()); + coinMapper.updateById(coin); + coinService.clearCache(coin.getCode()); + + // 日线收盘后清理 Redis 中的价格(下次 tick 会重新写入) + try { + redisTemplate.delete(REDIS_PRICE_PREFIX + coin.getCode().toUpperCase()); + } catch (Exception ignored) {} + + log.info("日线收盘更新: {} price={} change={}%", coin.getCode(), coin.getPrice(), change); + } + + private CoinKline copyKline(CoinKline source) { + CoinKline copy = new CoinKline(); + copy.setId(source.getId()); + copy.setCoinCode(source.getCoinCode()); + copy.setInterval(source.getInterval()); + copy.setOpenTime(source.getOpenTime()); + copy.setOpenPrice(source.getOpenPrice()); + copy.setHighPrice(source.getHighPrice()); + copy.setLowPrice(source.getLowPrice()); + copy.setClosePrice(source.getClosePrice()); + copy.setVolume(source.getVolume()); + copy.setCloseTime(source.getCloseTime()); + return copy; + } +} diff --git a/src/main/java/com/it/rattan/monisuo/service/KlineScheduler.java b/src/main/java/com/it/rattan/monisuo/service/KlineScheduler.java new file mode 100644 index 0000000..ad6c468 --- /dev/null +++ b/src/main/java/com/it/rattan/monisuo/service/KlineScheduler.java @@ -0,0 +1,72 @@ +package com.it.rattan.monisuo.service; + +import com.it.rattan.monisuo.dto.KlineTick; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.messaging.simp.SimpMessagingTemplate; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import java.util.List; + +/** + * K线调度器 — 定时生成 tick + 检测周期边界 + */ +@Slf4j +@Service +public class KlineScheduler { + + @Autowired + private CoinKlineService coinKlineService; + + @Autowired(required = false) + private SimpMessagingTemplate messagingTemplate; + + /** 上次 tick 的时间戳,用于检测周期边界跨越 */ + private long lastTickTime = System.currentTimeMillis(); + + /** + * 每 3 秒执行一次:生成价格 tick + WebSocket 广播 + */ + @Scheduled(fixedRate = 3000) + public void tick() { + try { + long now = System.currentTimeMillis(); + + // 1. 检测周期边界跨越 + List crossedIntervals = coinKlineService.detectIntervalCrossings(lastTickTime); + if (!crossedIntervals.isEmpty()) { + for (String interval : crossedIntervals) { + log.info("K线周期跨越: {}", interval); + List closedTicks = coinKlineService.closeCandlesForInterval(interval); + // 广播收盘 tick + for (KlineTick tick : closedTicks) { + broadcastTick(tick); + } + } + } + + // 2. 生成新的价格 tick + List ticks = coinKlineService.generateTicks(); + + // 3. WebSocket 广播 + for (KlineTick tick : ticks) { + broadcastTick(tick); + } + + lastTickTime = now; + } catch (Exception e) { + log.error("K线 tick 调度异常", e); + } + } + + private void broadcastTick(KlineTick tick) { + if (messagingTemplate != null) { + try { + messagingTemplate.convertAndSend("/topic/kline/" + tick.getCoinCode(), tick); + } catch (Exception e) { + log.warn("WebSocket 广播失败: {}", e.getMessage()); + } + } + } +}