第3章：核心概念解析
📖 章节概述
欢迎来到第3章！在这一章中，我们将深入理解Kronos的核心技术概念。掌握这些概念将帮助您更好地使用Kronos，并为其高级功能的应用打下坚实基础。
⏱️ 预计学习时间：3小时
🎯 学习目标：理解Kronos的技术原理和核心机制
📋 主要内容：K线数据语言化、分层量化技术、自回归预测、概率采样
3.1 K线数据语言化
3.1.1 从连续到离散的思维转变
传统金融时间序列模型通常直接处理连续的价格数据，而Kronos采用了类似自然语言处理的方法——将连续的金融数据"语言化"为离散的token序列。
🔄 传统方法 vs Kronos方法
graph LR
subgraph "传统方法"
A1[连续OHLCV数据] --> A2[数值计算模型] --> A3[连续预测输出]
end
subgraph "Kronos方法"
B1[连续OHLCV数据] --> B2[数据分词器] --> B3[离散Token序列] --> B4[语言模型] --> B5[预测Token] --> B6[反量化器] --> B7[连续预测输出]
end
💡 为什么选择"语言化"？
- 模式识别能力：语言模型擅长捕捉序列中的长期依赖关系
- 泛化能力：学会了金融市场的"通用语言"
- 可扩展性：可以处理不同类型和频率的金融数据
- 不确定性建模：天然支持概率预测
3.1.2 K线数据的特点与挑战
📊 OHLCV数据结构
import pandas as pd
import numpy as np
# 示例K线数据结构
def create_sample_kline_data():
dates = pd.date_range('2024-01-01', periods=100, freq='1H')
# 模拟价格走势
base_price = 100
returns = np.random.normal(0, 0.02, 100)
prices = base_price * (1 + returns).cumprod()
data = {
'open': prices,
'high': prices * np.random.uniform(1.0, 1.05, 100),
'low': prices * np.random.uniform(0.95, 1.0, 100),
'close': np.roll(prices, -1),
'volume': np.random.randint(1000, 10000, 100)
}
df = pd.DataFrame(data, index=dates)
return df
kline_data = create_sample_kline_data()
print("K线数据示例:")
print(kline_data.head())
🎯 主要挑战
- 多维度关联：开盘价、最高价、最低价、收盘价之间存在约束关系
- 时间依赖性：当前价格受历史价格影响
- 噪声干扰：市场包含大量随机波动
- 非平稳性：数据分布随时间变化
3.1.3 语言化处理的数学原理
📐 数据标准化
def normalize_kline_data(df):
"""K线数据标准化处理"""
# 计算对数收益率
log_returns = np.log(df['close'] / df['open'])
# Z-score标准化
normalized_data = (log_returns - log_returns.mean()) / log_returns.std()
return normalized_data
# 可视化标准化效果
import matplotlib.pyplot as plt
def plot_normalization_effect(original_data, normalized_data):
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
# 原始数据
ax1.plot(original_data.index, original_data['close'])
ax1.set_title('原始价格数据')
ax1.set_ylabel('价格')
# 标准化数据
ax2.plot(normalized_data.index, normalized_data)
ax2.set_title('标准化后的收益率')
ax2.set_ylabel('标准化值')
plt.tight_layout()
plt.show()
# 演示标准化过程
normalized_data = normalize_kline_data(kline_data)
plot_normalization_effect(kline_data, normalized_data)
3.2 分层量化技术（Binary Spherical Quantization）
3.2.1 量化技术概述
分层量化是Kronos的核心技术创新，它将连续的金融数据转换为离散的token表示，同时尽可能保留原始数据的统计特性和结构信息。
🎯 量化目标
- 信息保留：在离散化过程中最大程度保留有用信息
- 计算效率：离散token处理比连续数据更高效
- 结构保持：维持数据间的相对关系和约束
3.2.2 技术原理详解
📐 球面量化（Spherical Quantization）
球面量化将多维数据投影到单位球面上，然后进行离散化：
import numpy as np
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
def spherical_quantization_demo():
"""演示球面量化过程"""
# 生成2D示例数据
np.random.seed(42)
data = np.random.multivariate_normal([0, 0], [[1, 0.5], [0.5, 1]], 1000)
# 投影到单位圆
norms = np.linalg.norm(data, axis=1, keepdims=True)
unit_sphere_data = data / norms
# 角度离散化
angles = np.arctan2(unit_sphere_data[:, 1], unit_sphere_data[:, 0])
n_levels = 16
quantized_angles = np.round(angles / (2 * np.pi / n_levels)) * (2 * np.pi / n_levels)
# 可视化
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
# 原始数据
axes[0].scatter(data[:, 0], data[:, 1], alpha=0.6)
axes[0].set_title('原始数据')
axes[0].set_aspect('equal')
# 球面投影
axes[1].scatter(unit_sphere_data[:, 0], unit_sphere_data[:, 1], alpha=0.6)
axes[1].set_title('球面投影')
axes[1].set_aspect('equal')
# 量化结果
quantized_points = np.column_stack([
np.cos(quantized_angles),
np.sin(quantized_angles)
])
axes[2].scatter(quantized_points[:, 0], quantized_points[:, 1], alpha=0.6)
axes[2].set_title('量化结果')
axes[2].set_aspect('equal')
plt.tight_layout()
plt.show()
spherical_quantization_demo()
🔢 二进制编码（Binary Encoding）
量化后的角度值进一步转换为二进制编码：
def binary_encoding_demo():
"""演示二进制编码过程"""
n_levels = 16
n_bits = int(np.log2(n_levels))
print(f"使用 {n_bits} 位二进制数表示 {n_levels} 个级别\n")
for i in range(n_levels):
binary_str = format(i, f'0{n_bits}b')
angle = i * (2 * np.pi / n_levels)
print(f"级别 {i:2d}: 二进制 {binary_str} -> 角度 {angle:.3f} rad")
binary_encoding_demo()
3.2.3 完整量化流程实现
🏗️ 简化的量化器实现
import numpy as np
import torch
import torch.nn as nn
class SimplifiedKronosTokenizer(nn.Module):
"""简化的Kronos分词器实现"""
def __init__(self, n_levels=256, d_model=512):
super().__init__()
self.n_levels = n_levels
self.d_model = d_model
# 投影层
self.input_projection = nn.Linear(5, d_model) # OHLCV = 5维
# 量化层
self.quantizer = nn.Linear(d_model, n_levels)
# Token嵌入
self.token_embedding = nn.Embedding(n_levels, d_model)
def forward(self, x):
"""
x: (batch_size, seq_len, 5) OHLCV数据
返回: (batch_size, seq_len) token_ids
"""
# 投影到高维空间
projected = self.input_projection(x)
# 计算量化分数
quantization_scores = self.quantizer(projected)
# 获取token IDs
token_ids = torch.argmax(quantization_scores, dim=-1)
return token_ids
def decode(self, token_ids):
"""将token IDs解码回连续空间"""
token_embeddings = self.token_embedding(token_ids)
return token_embeddings
# 演示分词器使用
def demo_tokenizer():
"""演示分词器的使用"""
# 创建简化的分词器
tokenizer = SimplifiedKronosTokenizer(n_levels=64, d_model=128)
# 创建示例数据
batch_size, seq_len = 2, 10
sample_data = torch.randn(batch_size, seq_len, 5) # OHLCV
# 分词
token_ids = tokenizer(sample_data)
print(f"输入数据形状: {sample_data.shape}")
print(f"Token IDs形状: {token_ids.shape}")
print(f"Token IDs示例:\n{token_ids[0]}")
# 解码
decoded = tokenizer.decode(token_ids)
print(f"解码后形状: {decoded.shape}")
demo_tokenizer()
3.2.4 量化效果评估
📊 信息损失分析
def evaluate_quantization_quality(original_data, reconstructed_data):
"""评估量化质量"""
# 计算重构误差
mse = np.mean((original_data - reconstructed_data) ** 2)
# 计算相关性
correlation = np.corrcoef(original_data.flatten(), reconstructed_data.flatten())[0, 1]
# 计算分布相似性
from scipy import stats
ks_statistic, p_value = stats.ks_2samp(
original_data.flatten(),
reconstructed_data.flatten()
)
print(f"量化质量评估:")
print(f" 均方误差 (MSE): {mse:.6f}")
print(f" 相关系数: {correlation:.4f}")
print(f" KS统计量: {ks_statistic:.4f}")
print(f" p值: {p_value:.4f}")
return {
'mse': mse,
'correlation': correlation,
'ks_statistic': ks_statistic,
'p_value': p_value
}
# 模拟评估过程
def demo_quantization_evaluation():
"""演示量化效果评估"""
# 生成测试数据
np.random.seed(42)
original_data = np.random.randn(1000, 5)
# 模拟量化-重构过程
# 这里使用简化的量化过程进行演示
quantized_data = np.round(original_data * 10) / 10 # 简单量化
reconstructed_data = quantized_data # 简单重构
# 评估质量
metrics = evaluate_quantization_quality(original_data, reconstructed_data)
return metrics
demo_quantization_evaluation()
3.3 自回归预测机制
3.3.1 自回归模型基础
自回归（Autoregressive）模型是一种时间序列预测方法，它假设当前值可以由历史值的线性组合来预测。
📈 数学原理
对于时间序列 $y_t$，p阶自回归模型AR(p)定义为：
$$y_t = c + \sum_{i=1}^{p} \phi_i y_{t-i} + \epsilon_t$$
其中：
- $y_t$ 是时间 t 的值
- $c$ 是常数项
- $\phi_i$ 是模型参数
- $\epsilon_t$ 是误差项
🔄 Transformer中的自回归机制
Kronos使用Transformer架构实现自回归预测：
import torch
import torch.nn as nn
import math
class PositionalEncoding(nn.Module):
"""位置编码"""
def __init__(self, d_model, max_len=5000):
super().__init__()
pe = torch.zeros(max_len, d_model)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(torch.arange(0, d_model, 2).float() *
(-math.log(10000.0) / d_model))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
pe = pe.unsqueeze(0).transpose(0, 1)
self.register_buffer('pe', pe)
def forward(self, x):
return x + self.pe[:x.size(0), :]
class SimpleAutoregressiveTransformer(nn.Module):
"""简化的自回归Transformer模型"""
def __init__(self, vocab_size, d_model=512, nhead=8, num_layers=6):
super().__init__()
self.d_model = d_model
# Token嵌入
self.token_embedding = nn.Embedding(vocab_size, d_model)
self.pos_encoding = PositionalEncoding(d_model)
# Transformer层
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model,
nhead=nhead,
dim_feedforward=d_model * 4
)
self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
# 输出层
self.output_projection = nn.Linear(d_model, vocab_size)
def forward(self, x, mask=None):
"""
x: (seq_len, batch_size) token序列
"""
# 嵌入和位置编码
embedded = self.token_embedding(x) * math.sqrt(self.d_model)
embedded = self.pos_encoding(embedded)
# Transformer处理
transformed = self.transformer(embedded, src_key_padding_mask=mask)
# 输出投影
output = self.output_projection(transformed)
return output
def generate(self, start_tokens, max_len=100, temperature=1.0):
"""自回归生成"""
self.eval()
with torch.no_grad():
current_seq = start_tokens.clone()
generated = []
for _ in range(max_len):
# 获取下一个token的预测
output = self.forward(current_seq)
next_token_logits = output[-1, :] / temperature
# 采样下一个token
next_token = torch.multinomial(
torch.softmax(next_token_logits, dim=-1),
num_samples=1
)
generated.append(next_token)
current_seq = torch.cat([current_seq, next_token.unsqueeze(0)], dim=0)
return torch.cat(generated)
# 演示自回归生成
def demo_autoregressive_generation():
"""演示自回归生成过程"""
# 创建模型
model = SimpleAutoregressiveTransformer(vocab_size=1000, d_model=128, num_layers=2)
# 创建起始序列
start_tokens = torch.randint(0, 1000, (10, 1)) # (seq_len, batch_size)
print("起始token序列:", start_tokens.squeeze().tolist())
# 生成新tokens
generated_tokens = model.generate(start_tokens, max_len=5, temperature=1.0)
print("生成的token序列:", generated_tokens.tolist())
return model, generated_tokens
demo_autoregressive_generation()
3.3.2 因果注意力机制
为了确保自回归性质，Transformer使用因果注意力（Causal Attention），确保每个位置只能注意到之前的位置：
def generate_causal_mask(seq_len):
"""生成因果注意力掩码"""
mask = torch.triu(torch.ones(seq_len, seq_len), diagonal=1).bool()
return mask
def demo_causal_attention():
"""演示因果注意力机制"""
seq_len = 5
mask = generate_causal_mask(seq_len)
print("因果注意力掩码:")
print("True表示被遮蔽的位置")
for i in range(seq_len):
row = []
for j in range(seq_len):
if mask[i, j]:
row.append("X") # 被遮蔽
else:
row.append("●") # 可见
print(f"位置{i}: {' '.join(row)}")
return mask
demo_causal_attention()
3.3.3 多步预测策略
🎯 递归预测（Recursive Prediction）
def recursive_prediction(model, initial_sequence, n_steps):
"""递归多步预测"""
model.eval()
current_sequence = initial_sequence.clone()
predictions = []
with torch.no_grad():
for step in range(n_steps):
# 预测下一个时间点
next_pred = model(current_sequence.unsqueeze(0)) # 添加batch维度
next_token = torch.argmax(next_pred[0, -1, :])
predictions.append(next_token)
# 更新序列
current_sequence = torch.cat([
current_sequence[1:], # 移除最早的token
next_token.unsqueeze(0) # 添加新预测的token
])
return torch.stack(predictions)
# 演示递归预测
def demo_recursive_prediction():
"""演示递归预测过程"""
# 创建简化模型
model = SimpleAutoregressiveTransformer(vocab_size=100, d_model=64, num_layers=2)
# 初始序列
initial_seq = torch.randint(0, 100, (10,))
print("初始序列:", initial_seq.tolist())
# 递归预测
predictions = recursive_prediction(model, initial_seq, n_steps=5)
print("预测结果:", predictions.tolist())
return predictions
demo_recursive_prediction()
🎯 直接预测（Direct Prediction）
def direct_prediction(model, input_sequence, n_steps):
"""直接多步预测"""
model.eval()
with torch.no_grad():
# 一次性预测多个未来时间点
output = model(input_sequence.unsqueeze(0)) # (1, seq_len, vocab_size)
# 取最后n_steps个预测
multi_step_preds = output[0, -n_steps:, :]
predicted_tokens = torch.argmax(multi_step_preds, dim=-1)
return predicted_tokens
# 比较两种预测策略
def compare_prediction_strategies():
"""比较递归预测和直接预测"""
model = SimpleAutoregressiveTransformer(vocab_size=100, d_model=64, num_layers=2)
# 输入序列
input_seq = torch.randint(0, 100, (15,))
n_steps = 5
print("输入序列长度:", len(input_seq))
print("预测步数:", n_steps)
# 递归预测
recursive_preds = recursive_prediction(model, input_seq[:10], n_steps)
print("递归预测:", recursive_preds.tolist())
# 直接预测
direct_preds = direct_prediction(model, input_seq, n_steps)
print("直接预测:", direct_preds.tolist())
return recursive_preds, direct_preds
compare_prediction_strategies()
3.4 概率采样与不确定性
3.4.1 预测不确定性的来源
金融预测中的不确定性主要来源于：
- 模型不确定性：模型结构或参数的不确定性
- 数据不确定性：数据本身的噪声和随机性
- 市场不确定性：外部环境和市场结构的变化
3.4.2 温度采样（Temperature Sampling）
温度采样通过调整概率分布的"尖锐度"来控制预测的随机性：
🌡️ 温度参数的作用
def temperature_sampling(logits, temperature=1.0):
"""温度采样"""
# 调整logits
scaled_logits = logits / temperature
# 计算概率分布
probabilities = torch.softmax(scaled_logits, dim=-1)
# 采样
samples = torch.multinomial(probabilities, num_samples=1)
return samples, probabilities
def demo_temperature_effect():
"""演示温度参数对采样的影响"""
# 示例logits
logits = torch.tensor([1.0, 2.0, 3.0, 0.5, 1.5])
temperatures = [0.1, 0.5, 1.0, 2.0, 5.0]
print("温度参数对采样的影响:")
print(f"原始logits: {logits.tolist()}")
print()
for temp in temperatures:
samples, probs = temperature_sampling(logits, temperature=temp)
print(f"温度 {temp:4.1f}:")
print(f" 概率分布: {probs.numpy().round(3)}")
print(f" 采样结果: {samples.item()}")
print(f" 熵: {-(probs * torch.log(probs + 1e-8)).sum():.3f}")
print()
demo_temperature_effect()
📊 温度参数选择指南
| 温度值 | 特点 | 适用场景 |
|---|---|---|
| 0.1 - 0.3 | 预测确定性高，选择最可能的token | 保守策略，高风险规避 |
| 0.5 - 1.0 | 平衡确定性和多样性 | 一般预测场景 |
| 1.0 - 2.0 | 增加随机性，探索更多可能性 | 创新策略，高风险高回报 |
| 2.0+ | 高度随机，接近均匀分布 | 压力测试，极端情况分析 |
3.4.3 核采样（Nucleus Sampling）
核采样选择累积概率超过阈值的最小token集合：
def nucleus_sampling(logits, top_p=0.9):
"""核采样实现"""
# 按概率降序排列
sorted_logits, sorted_indices = torch.sort(logits, descending=True)
sorted_probs = torch.softmax(sorted_logits, dim=-1)
# 计算累积概率
cumulative_probs = torch.cumsum(sorted_probs, dim=-1)
# 移除累积概率超过top_p的tokens
sorted_indices_to_remove = cumulative_probs > top_p
sorted_indices_to_remove[1:] = sorted_indices_to_remove[:-1].clone()
sorted_indices_to_remove[0] = 0
# 创建mask并应用到原始logits
indices_to_remove = sorted_indices_to_remove.scatter(0, sorted_indices, sorted_indices_to_remove)
filtered_logits = logits.clone()
filtered_logits[indices_to_remove] = float('-inf')
# 重新计算概率并采样
final_probs = torch.softmax(filtered_logits, dim=-1)
samples = torch.multinomial(final_probs, num_samples=1)
return samples, final_probs
def demo_nucleus_sampling():
"""演示核采样效果"""
# 示例：多个候选token
logits = torch.tensor([2.0, 1.8, 1.5, 1.2, 1.0, 0.8, 0.5, 0.3])
top_p_values = [0.5, 0.7, 0.9, 0.95]
print("核采样效果演示:")
print(f"原始logits: {logits.tolist()}")
print()
for top_p in top_p_values:
samples, probs = nucleus_sampling(logits, top_p=top_p)
# 统计被保留的token数量
n_selected = (probs > 0).sum().item()
print(f"Top-P {top_p:.2f}:")
print(f" 保留token数: {n_selected}/8")
print(f" 最终概率: {probs.numpy().round(3)}")
print(f" 采样结果: {samples.item()}")
print()
demo_nucleus_sampling()
3.4.4 多样本预测与置信度估计
🎲 多样本预测
def multi_sample_prediction(model, input_sequence, n_samples=10, temperature=1.0, top_p=0.9):
"""多样本预测"""
model.eval()
predictions = []
with torch.no_grad():
for _ in range(n_samples):
# 获取模型输出
output = model(input_sequence.unsqueeze(0))
last_logits = output[0, -1, :]
# 应用温度和核采样
if top_p < 1.0:
sample, _ = nucleus_sampling(last_logits, top_p=top_p)
else:
sample, _ = temperature_sampling(last_logits, temperature=temperature)
predictions.append(sample.item())
return predictions
def calculate_prediction_confidence(predictions):
"""计算预测置信度"""
from collections import Counter
# 统计预测分布
counter = Counter(predictions)
total_samples = len(predictions)
# 计算概率分布
prob_distribution = {k: v/total_samples for k, v in counter.items()}
# 计算熵（不确定性度量）
import math
entropy = -sum(p * math.log(p) for p in prob_distribution.values())
# 计算最频繁预测的置信度
most_common = counter.most_common(1)[0]
confidence = most_common[1] / total_samples
return {
'predictions': predictions,
'distribution': prob_distribution,
'entropy': entropy,
'confidence': confidence,
'most_common': most_common
}
def demo_uncertainty_quantification():
"""演示不确定性量化"""
# 创建模型
model = SimpleAutoregressiveTransformer(vocab_size=50, d_model=64, num_layers=2)
# 输入序列
input_seq = torch.randint(0, 50, (10,))
# 多样本预测
predictions = multi_sample_prediction(
model, input_seq,
n_samples=100,
temperature=1.0,
top_p=0.9
)
# 计算置信度
results = calculate_prediction_confidence(predictions)
print("不确定性量化结果:")
print(f" 样本数: {len(predictions)}")
print(f" 预测分布: {results['distribution']}")
print(f" 熵 (不确定性): {results['entropy']:.3f}")
print(f" 置信度: {results['confidence']:.3f}")
print(f" 最常见预测: token {results['most_common'][0]} ({results['most_common'][1]}次)")
return results
demo_uncertainty_quantification()
3.4.5 实际应用中的不确定性管理
📊 预测结果可视化
def visualize_prediction_uncertainty(predictions, true_value=None):
"""可视化预测不确定性"""
import matplotlib.pyplot as plt
# 统计预测分布
from collections import Counter
counter = Counter(predictions)
# 创建图表
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
# 预测分布直方图
tokens = list(counter.keys())
counts = list(counter.values())
ax1.bar(tokens, counts, alpha=0.7)
if true_value is not None:
ax1.axvline(true_value, color='red', linestyle='--', label=f'真实值: {true_value}')
ax1.legend()
ax1.set_xlabel('Token ID')
ax1.set_ylabel('预测次数')
ax1.set_title('预测分布')
# 概率分布
total = sum(counts)
probs = [c/total for c in counts]
ax2.bar(tokens, probs, alpha=0.7)
if true_value is not None and true_value in tokens:
idx = tokens.index(true_value)
ax2.bar([true_value], [probs[idx]], color='red', alpha=0.7, label='真实值概率')
ax2.legend()
ax2.set_xlabel('Token ID')
ax2.set_ylabel('预测概率')
ax2.set_title('概率分布')
plt.tight_layout()
plt.show()
# 演示不确定性可视化
def demo_uncertainty_visualization():
"""演示不确定性可视化"""
# 模拟预测结果
np.random.seed(42)
predictions = np.random.choice([10, 11, 12, 13, 14, 15], size=100, p=[0.1, 0.15, 0.4, 0.2, 0.1, 0.05])
true_value = 12
visualize_prediction_uncertainty(predictions, true_value)
demo_uncertainty_visualization()
3.5 章节小结
🎯 核心要点回顾
通过本章学习，您应该掌握：
- K线数据语言化：理解为什么将金融数据转换为离散token序列
- 分层量化技术：掌握球面量化和二进制编码的原理
- 自回归预测：了解Transformer如何实现时间序列预测
- 概率采样：学会使用温度和核采样处理预测不确定性
💡 重要概念
- Tokenization：将连续数据转换为离散token的过程
- Spherical Quantization：保持数据结构的球面投影量化方法
- Autoregressive Prediction：基于历史序列预测未来的方法
- Temperature Sampling：控制预测随机性的技术
- Uncertainty Quantification：评估预测置信度的方法
🔧 实用技能
- 实现简单的数据量化算法
- 使用因果注意力进行序列预测
- 应用温度和核采样生成多样化预测
- 计算和可视化预测不确定性
Top comments (0)