让K线说话！用形态匹配功能透视通达信数据黑洞
摘要：磐石数据处理系统新增跨平台支持（Windows/统信UOS）、通达信日线数据导入及K线形态匹配功能。系统可通过框选5-25根K线进行形态匹配，计算相似度并统计未来上涨概率。Windows版内置Python环境，统信版需手动安装依赖包。核心功能通过Python实现，包括数据标准化处理、滑动窗口相似度计算（使用dtaidistance库）及未来涨幅预测。系统提供图形化操作界面，支持K线缩放平移，
目录
1 简介
近期为数据处理系统增加了部分功能。主要包括：
1、增加了跨平台功能。当前有windows版本和统信UOS版本（aarch64-linux）。使用秋风版Lazarus，可以很方便地编译生成各种版本应用程序；
2、数据库中导入通达信日线数据；
3、画K线图，选取特定日期，从历史记录中匹配查找形态类似K线，并统计类似K先后股票上涨概率情况；
4、画出匹配出的K线，并标记出选定日期。
再就是，程序更名为“磐石数据处理系统”。
2 下载
Windows版本和统信UOS（aarch64-linux）版本都在一个压缩包里面。
通过网盘分享的文件：panstone-20250717.zip
链接: https://pan.baidu.com/s/1VhPZ-7RC70oYiSGygu1Quw?pwd=hpbs 提取码: hpbs
以下是简单操作步骤：
3 通达信日线数据导入
菜单位置：金融——导入通达信日线数据
选好通达信程序安装文件夹，切换一下市场，或者重启程序，就可以看到股票列表。
全选或者部分选择（按住ctrl或者shift+鼠标点击）股票，点击“导入”。
如果选择太多股票，会很费时间。
4 形态匹配
菜单位置：金融——数据分析
4.1 设置形态最低相似度和最大返回结果数量
定义最低相似度，返回数量。
最低匹配度越低，返回数量越多，越费时间。
4.2 框选操作
双击一个股票名称，显示K线图。
在k线图中框选一部分k线，可以放大框选部分。
4.3 放大或缩小
注意框选的时候从左向右选，放大K线图，反之为缩小。
选出5-25根K线，形态匹配按钮可用，否则不可用。
4.4 K线图平移
用鼠标右键拖动K线图，可以实现平移。
4.5 匹配命令
点击“形态匹配”。
第一次点击有时不出结果，等一会后再点击就可以了。
匹配完成，左侧显示匹配结果列表，右侧出来一个简单统计数据。
双击列表中的一个代码，显示匹配K线图：
5 不同平台使用注意
5.1 Windows
windows版本已经内置python embedded版，并且安装完成所需的包，直接运行即可，但是体积有点大。
5.2 统信linux
该版本程序体积较小，但需要自行安装所需的python包。
5.3 统信系统安装python包的方法
首先在程序中运行某项功能，程序会信息页签列出本次执行脚本的完整命令。如：
可以在终端运行的命令: python3 D:\00panstone\py\stock_chart\stock_chart.py
000680.SH
2024-12-24
2024-12-30
data\data.db3
D:\00panstone\temp
需要把上述命令整理为一行：
python3 D:\00panstone\py\stock_chart\stock_chart.py 000680.SH 2024-12-24 2024-12-30 data\data.db3 D:\00panstone\temp然后复制到终端中运行，缺少哪个包，会有提示。根据提示安装包。以下是可能需要的安装命令：
1 更新软件源
udo apt update
如果提示失败，添加缺失的 GPG 公钥
udo apt-key adv --keyserver keyserver.ubuntu.com --recv-keys 3AF65F93D6FBC5B9
2 安装 Python 3 的 pip
在终端运行以下命令：
sudo apt install python3-pip
3. 验证安装
pip3 --version
4 使用 pip 安装 NumPy
pip3 install numpy
# 或者使用全局 pip
pip install numpy
pip3 install dtaidistance
pip3 install pandas
安装matplotlib失败后：
sudo apt update
sudo apt install build-essential python3-dev libjpeg-dev zlib1g-dev
sudo apt install libpng-dev libtiff-dev libwebp-dev libfreetype6-dev
pip3 install --upgrade pip setuptools
pip3 uninstall -y pillow matplotlib
rm -rf ~/.cache/pip
pip3 install --no-cache-dir matplotlib
6 搜索匹配形态的python函数代码
def find_similar_patterns(db_path, query_closes, threshold, res_max_cnt):
"""查找相似形态"""
global progress_data
conn = None
try:
logging.info("连接数据库: %s" % db_path)
update_progress("", 0, 0, "connecting to database")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# 获取所有股票代码
logging.info("获取所有股票代码...")
update_progress("", 0, 0, "fetching stock codes")
cursor.execute("SELECT DISTINCT code FROM tdx_stock_daily")
all_stocks = [row[0] for row in cursor.fetchall()]
total_stocks = len(all_stocks)
logging.info("找到 %s 只股票" % total_stocks)
# 更新总进度
update_progress("", 0, total_stocks, "starting processing")
# 标准化查询序列
norm_query = normalize_sequence(query_closes)
window_size = len(norm_query)
if window_size < 5:
return {"status": "error", "message": "查询序列太短（至少需要5个数据点）"}
if window_size > 25:
return {"status": "error", "message": "查询序列太长（最多支持25个数据点）"}
results = []
total_windows = 0
processed_stocks = 0
# 遍历所有股票
for stock_code in all_stocks:
processed_stocks += 1
# 每100只股票记录一次
if processed_stocks % 100 == 0:
logging.info("处理进度: %s/%s 只股票" % (processed_stocks, len(all_stocks)))
# 更新进度
update_progress(stock_code, processed_stocks, total_stocks, "processing")
# 获取股票数据
cursor.execute(
"SELECT trade_date, close FROM tdx_stock_daily "
"WHERE code = ? ORDER BY trade_date",
(stock_code,)
)
rows = cursor.fetchall()
closes = [row[1] for row in rows]
dates = [row[0] for row in rows]
# 如果数据不足，跳过
if len(closes) < window_size:
continue
# 滑动窗口遍历
for i in range(len(closes) - window_size + 1):
total_windows += 1
window = closes[i:i+window_size]
norm_window = normalize_sequence(window)
# 计算相似度
try:
similarity = calculate_similarity(norm_query, norm_window)
# 只添加相似度超过阈值的匹配项
if similarity >= threshold:
start_date = dates[i]
end_date = dates[i+window_size-1]
results.append({
'stock_code': stock_code,
'start_date': start_date,
'end_date': end_date,
'similarity': similarity
})
# 检查是否达到最大结果数量
if len(results) >= res_max_cnt:
logging.info("已达到最大结果数量(%d)，停止搜索" % res_max_cnt)
# 跳出内层循环
break
except Exception as e:
logging.error("计算相似度时出错 (股票: %s, 位置: %s): %s" % (stock_code, i, str(e)))
# 检查是否达到最大结果数量，跳出外层循环
if len(results) >= res_max_cnt:
break
# 处理完成
update_progress("", processed_stocks, total_stocks, "completed")
# 按相似度排序
results.sort(key=lambda x: x['similarity'], reverse=True)
# 记录匹配统计信息
match_rate = len(results) / total_windows * 100 if total_windows > 0 else 0
logging.info("找到 %d 个匹配形态 (总窗口数: %d, 匹配率: %.2f%%)" % (len(results), total_windows, match_rate))
# 计算5日内出现5%涨幅的几率
high_gain_count = 0
valid_results = 0
# 只处理找到的结果（不超过最大数量）
max_results_to_process = min(res_max_cnt, len(results))
logging.info("计算前%s个匹配形态的未来涨幅..." % max_results_to_process)
for i, result in enumerate(results[:max_results_to_process]):
stock_code = result['stock_code']
end_date = result['end_date']
# 计算未来5日最大涨幅
max_gain = calculate_future_gain(db_path, stock_code, end_date)
if max_gain is not None:
valid_results += 1
result['future_max_gain'] = max_gain
# 统计涨幅≥5%的情况
if max_gain >= 5.0:
high_gain_count += 1
# 计算几率
high_gain_probability = (high_gain_count / valid_results * 100) if valid_results > 0 else 0
logging.info("5日内涨幅≥5%%的几率: %.2f%% (%d/%d)" % (high_gain_probability, high_gain_count, valid_results))
return {
"status": "success",
"results": results[:res_max_cnt],
"count": len(results),
"match_rate": match_rate,
"total_windows": total_windows,
"high_gain_probability": high_gain_probability,
"high_gain_count": high_gain_count,
"valid_results": valid_results
}
except Exception as e:
# 错误处理
update_progress(stock_code if 'stock_code' in locals() else "",
processed_stocks if 'processed_stocks' in locals() else 0,
total_stocks,
"error")
logging.exception("查找相似形态时发生错误")
return {
"status": "error",
"message": str(e),
"traceback": traceback.format_exc()
}
finally:
if conn:
conn.close()
logging.info("数据库连接已关闭")
全面兼容主流 AI 模型，支持本地及云端双模式
更多推荐
9
0- 0
已为社区贡献1条内容
所有评论(0)