让K线说话！用形态匹配功能透视通达信数据黑洞
目录
1 简介
2 下载
3 通达信日线数据导入
4 形态匹配
4.1 设置形态最低相似度和最大返回结果数量
4.2 框选操作
4.3 放大或缩小
4.4 K线图平移
4.5 匹配命令
5 不同平台使用注意
5.1 Windows
5.2 统信linux
5.3 统信系统安装python包的方法
6 搜索匹配形态的python函数代码
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 3AF65F93D6FBC5B92 安装 Python 3 的 pip
在终端运行以下命令：sudo apt install python3-pip3. 验证安装
pip3 --version4 使用 pip 安装 NumPy
pip3 install numpy
# 或者使用全局 pip
pip install numpypip3 install dtaidistancepip3 install pandas安装matplotlib失败后：sudo apt update
sudo apt install build-essential python3-dev libjpeg-dev zlib1g-devsudo apt install libpng-dev libtiff-dev libwebp-dev libfreetype6-devpip3 install --upgrade pip setuptools
pip3 uninstall -y pillow matplotlib
rm -rf ~/.cache/pippip3 install --no-cache-dir matplotlib6 搜索匹配形态的python函数代码
def find_similar_patterns(db_path, query_closes, threshold, res_max_cnt):"""查找相似形态"""global progress_dataconn = Nonetry: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 = 0processed_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 += 1window = 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)# 跳出内层循环breakexcept 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 0logging.info("找到 %d 个匹配形态 (总窗口数: %d, 匹配率: %.2f%%)" % (len(results), total_windows, match_rate))# 计算5日内出现5%涨幅的几率high_gain_count = 0valid_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 += 1result['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 0logging.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("数据库连接已关闭")