
等级: 专业版
- 注册:
- 2021-10-19
- 曾用名:
|
金字塔里面不是有个功能 就是PEL引用PYTHON 模块吗? 我做了以下尝试 但是系统写出来的代码每一行都提示函数未定义
被引用的 PYTHON 的文件名为:MyPython9 具体代码如下:
# 该 Python 代码用于提供高效的订单流数据计算
import numpy as np
from datetime import datetime
def init(context):
context.N = 1 # 周期参数
context.num_levels = 30 # 显示价格档位数
context.min_diff = 0.01 # 最小变动单位
context.merge_level = 1 # 合并价位数
def handle_bar(context):
# 获取配置参数
N, levels, tick_size = context.N, context.num_levels, context.min_diff
复制
# 智能获取有效周期
period = calculate_valid_period(context, N)
if period == 0: return
# 获取历史数据并转换为结构化数组
bars = pel_history_bars(period, ['datetime', 'close', 'buyvol', 'sellvol'])
if len(bars) == 0: return
# 生成价格矩阵
current_price = round(bars[-1]['close'], 2)
price_levels = generate_price_matrix(current_price, levels, tick_size)
# 使用向量化计算买卖量
buy_sell = calculate_volume(bars, price_levels, tick_size)
# 动态设置属性
set_attributes(context, price_levels, buy_sell)
def calculate_valid_period(context, N):
"""智能计算有效周期"""
try:
time_series = pel_history_bars(1440, 'datetime') # 获取当日数据
if len(time_series) < 2: return 1
复制
current_dt = parse_datetime(time_series[-1])
time_diff = [current_dt - parse_datetime(ts) for ts in time_series[:-1]]
# 寻找最近 N分钟边界
threshold = N * 60 # 转换为秒
for i, td in enumerate(reversed(time_diff)):
if td.total_seconds() > threshold:
return i + 1
return len(time_series)
except:
return 1 # 异常时返回默认值
def parse_datetime(ts):
"""高效时间解析"""
return datetime.strptime(f"{ts:014}", "%Y%m%d%H%M%S")
def generate_price_matrix(base_price, levels, tick):
"""生成价格矩阵"""
prices = {}
prices['价现 0'] = round(base_price, 2)
for i in range(1, levels+1):
prices[f'价上{i}'] = round(base_price + itick, 2)
prices[f'价下{i}'] = round(base_price - itick, 2)
return prices
def calculate_volume(bars, price_levels, tick):
"""向量化计算成交量"""
# 转换为 NumPy 数组
closes = np.round([bar['close'] for bar in bars], decimals=2)
buyvol = np.array([bar.get('buyvol',0) for bar in bars])
sellvol = np.a
|
|