from __future__ import annotations from collections import OrderedDict from .state import DEFAULT_TIMEFRAME_LABELS def _zone_cache_ttl(tf_name: str) -> int: """根据时间周期返回缓存过期时间(秒)""" minutes = timeframe_to_minutes(tf_name) or 5 if minutes <= 5: return 120 # 5m及以下: 2分钟 elif minutes <= 15: return 300 # 15m: 5分钟 elif minutes <= 60: return 600 # 1h: 10分钟 else: return 1800 # 4h+: 30分钟 def timeframe_to_minutes(tf: str): """将时间周期转换为分钟数,用于排序。""" if not tf: return None unit = tf[-1] try: value = int(tf[:-1]) except (ValueError, TypeError): return None multiplier = { 'm': 1, 'h': 60, 'd': 1440, 'w': 10080, 'M': 43200, # 30天近似 }.get(unit) if multiplier is None: return None return value * multiplier def format_timeframe_label(tf: str) -> str: """将时间周期转换为可读标签。""" if not tf: return tf unit = tf[-1] try: value = int(tf[:-1]) except (ValueError, TypeError): return tf if unit == 'm': return f"{value}分钟" if unit == 'h': return f"{value}小时" if unit == 'd': return "日线" if value == 1 else f"{value}日线" if unit == 'w': return "周线" if value == 1 else f"{value}周线" if unit == 'M': return "月线" if value == 1 else f"{value}月线" return tf def build_timeframe_labels(timeframes): ordered = sorted( timeframes, key=lambda tf: timeframe_to_minutes(tf) if timeframe_to_minutes(tf) is not None else float('inf'), ) labels = OrderedDict() for tf in ordered: labels[tf] = format_timeframe_label(tf) return labels def _adjacent_smaller(timeframe_keys, ceiling_tf): """取排序列表中严格小于 ceiling 的相邻周期。""" if not timeframe_keys: return ceiling_tf try: idx = timeframe_keys.index(ceiling_tf) return timeframe_keys[idx - 1] if idx > 0 else timeframe_keys[0] except ValueError: return timeframe_keys[0] def _prefer_smaller(candidates, labels_ordered, ceiling_tf, timeframe_keys): """从候选中选第一个存在且严格小于 ceiling 的周期,否则回退相邻更小。""" ceil_m = timeframe_to_minutes(ceiling_tf) for tf in candidates: m = timeframe_to_minutes(tf) if tf in labels_ordered and m is not None and ceil_m is not None and m < ceil_m: return tf return _adjacent_smaller(timeframe_keys, ceiling_tf) def compute_timeframe_defaults(labels_ordered): """ 根据已排序的「周期 → 中文标签」映射,计算主 / 次 / 次次周期默认值。 默认偏好:主 4h、次 1h、次次 15m。 labels_ordered: OrderedDict 或按插入顺序排列的 dict。 """ if not labels_ordered: labels_ordered = DEFAULT_TIMEFRAME_LABELS.copy() timeframe_keys = list(labels_ordered.keys()) preferred_main = next((tf for tf in ['4h', '1h', '15m'] if tf in labels_ordered), None) default_main = preferred_main or (timeframe_keys[0] if timeframe_keys else '1m') if default_main not in labels_ordered and timeframe_keys: default_main = timeframe_keys[0] default_element = _prefer_smaller(['1h', '15m'], labels_ordered, default_main, timeframe_keys) default_sub_sub = _prefer_smaller(['15m', '5m'], labels_ordered, default_element, timeframe_keys) return default_main, default_element, default_sub_sub, timeframe_keys def is_smaller_timeframe(tf1, tf2): """判断时间周期tf1是否小于tf2""" tf1_value = timeframe_to_minutes(tf1) tf2_value = timeframe_to_minutes(tf2) if tf1_value is None or tf2_value is None: return False return tf1_value < tf2_value def is_smaller_or_equal_timeframe(tf1, tf2): """判断时间周期tf1是否小于等于tf2""" tf1_value = timeframe_to_minutes(tf1) tf2_value = timeframe_to_minutes(tf2) if tf1_value is None or tf2_value is None: return False return tf1_value <= tf2_value