from jqdata import * def initialize(context): # 策略初始化设置 set_params(context) # 设置参数 set_backtest() # 设置回测条件 def set_params(context): # 配置策略参数 context.stock_pool = [ {'code': '399673.XSHE', 'name': '创业板50'}, {'code': '000688.XSHG', 'name': '科创50'}, {'code': '399303.XSHE', 'name': '国证2000'} ] context.bond_code = '000012.XSHG' # 国债指数 context.cy_index = '399006.XSHE' # 创业板指 context.ma_days = 20 # 均线周期 context.rank_days = 20 # 排名计算周期 context.sell_ma = 10 # 卖出均线周期 def set_backtest(): # 回测设置 set_benchmark('399673.XSHE') set_option('use_real_price', True) set_order_cost(OrderCost( open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5 ), type='stock') def handle_data(context, data): # 每日交易逻辑 if not is_trading_day(context): return check_sell_condition(context, data) # 持仓时判断平仓 check_buy_condition(context, data) # 空仓时判断开仓 def is_trading_day(context): # 判断是否为交易日 return not (context.current_dt.hour == 15 and context.current_dt.minute == 0) def check_sell_condition(context, data): # 平仓条件判断 for stock in context.portfolio.positions: if stock.code == context.bond_code: continue current_price = data[stock.code].close ma10 = get_ma(stock.code, context.sell_ma, data) if current_price < ma10: clear_position(context, stock.code) log.info(f"触发平仓条件：{stock.code} 价格跌破10日均线") def check_buy_condition(context, data): # 开仓条件判断 if len(context.portfolio.positions) > 0: return # 大盘择时判断 cy_price = data[context.cy_index].close cy_ma20 = get_ma(context.cy_index, context.ma_days, data) if cy_price <= cy_ma20: return # 计算标的排名 ranked_stocks = [] for stock in context.stock_pool: current_price = data[stock['code']].close ma20 = get_ma(stock['code'], context.rank_days, data) ratio = (current_price - ma20) / ma20 ranked_stocks.append((stock, ratio)) # 按涨幅排序 ranked_stocks.sort(key=lambda x: x[1], reverse=True) target_stock = ranked_stocks[0][0] # 开仓条件验证 current_price = data[target_stock['code']] cy_ma20 = get_ma(context.cy_index, context.ma_days, data) NameError: name 'get_ma' is not defined

LK-GD500 initialize_labview基恩士参数初始化_基恩士_

在本教程中，我们将探讨“LK-GD500 initialize_labview基恩士参数初始化”这一主题，这涉及到使用LabVIEW（Laboratory Virtual Instrument Engineering Workbench）软件对基恩士LK-GD500激光传感器进行初始化设置和...

identity_cache_fetchers:用于 IdentityCache 的 ActiveRecord#fetch_or_initialize 和 ActiveRecord#fetch_or_create

身份缓存提取器用于ActiveRecord#fetch_or_initialize 和 ActiveRecord#fetch_or_create # based on https://github.com/Shopify/identity_cache#secondary-indexesclass Product <... fetch_or_initialize_by

LK-GD500-initialize.zip_LABVIEW 基恩士_LK-_LK-GD500设置_labview基恩士_恩基

基恩士LK-GD500激光感测初始化设置和参数的录入

from jqdata import * def initialize(context): set_params(context) # 设置参数 set_backtest() # 设置回测条件 def set_params(context): context.stock_pool = [ {'code': '399673.XSHE', 'name': '创业板50'}, {'code': '000688.XSHG', 'name': '科创50'}, {'code': '399303.XSHE', 'name': '国证2000'} ] context.bond_code = '000012.XSHG' # 国债指数 context.cy_index = '399006.XSHE' # 创业板指 context.ma_days = 20 # 均线周期 context.rank_days = 20 # 排名计算周期 context.sell_ma = 10 # 卖出均线周期 def set_backtest(): set_benchmark('399673.XSHE') set_option('use_real_price', True) set_order_cost(OrderCost( open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5 ), type='stock') # 新增移动平均计算函数 def get_ma(security, n_days, data): """计算N日移动平均线""" hist = attribute_history(security, n_days, fields='close') return hist['close'].mean() def handle_data(context, data): if not is_trading_day(context): return check_sell_condition(context, data) check_buy_condition(context, data) def is_trading_day(context): return not (context.current_dt.hour == 15 and context.current_dt.minute == 0) def check_sell_condition(context, data): for stock in context.portfolio.positions: if stock.code == context.bond_code: continue current_price = data[stock.code].close ma10 = get_ma(stock.code, context.sell_ma, data) if current_price < ma10: clear_position(context, stock.code) log.info(f"触发平仓条件：{stock.code} 价格跌破10日均线") def check_buy_condition(context, data): if len(context.portfolio.positions) > 0: return # 大盘择时判断 cy_price = data[context.cy_index].close cy_ma20 = get_ma(context.cy_index, context.ma_days, data) if cy_price <= cy_ma20: return # 计算标的排名 ranked_stocks = [] for stock in context.stock_pool: current_price = data[stock['code']].close ma20 = get_ma(stock['code'], context.rank_days, data) ratio = (current_price - ma20) / ma20 ranked_stocks.append((stock, ratio)) # 按涨幅排序 ranked_stocks.sort(key=lambda x: x[1], reverse=True) target_stock = ranked_stocks[0][0] # 修正此行语法错误 current_price = data[target_stock['code']].close # 获取标的当前价格 cy_ma20 = get_ma(context.cy_index, context.ma_days, data) # 重新获取创业板指MA20 # 最终开仓判断 if current_price > cy_ma20: order_value(target_stock['code'], context.portfolio.available_cash) log.info(f"开仓买入：{target_stock['name']}") def clear_position(context, code): """清空指定标的持仓""" order_target(code, 0) File "/tmp/strategy/user_code.py", line 38, in handle_data check_sell_condition(context, data) File "/tmp/strategy/user_code.py", line 46, in check_sell_condition if stock.code == context.bond_code: continue AttributeError: 'str' object has no attribute 'code'

1. **初始化设置** - 配置三只指数基金（创业板50、科创50、国证2000）作为股票池 - 使用国债指数000012.XSHG作为债券标的 - 设置均线周期为20日，卖出均线周期为10日 2. **交易逻辑** **卖出条件**（每日...

# 简化版IM1000期货策略 - 简单粗暴赚钱版 from jqdata import * import talib import numpy as np def initialize(context): # 设置全局变量 g.future_type = 'IM' # 交易IM期货 g.benchmark = '000852.XSHG' # 中证1000指数 g.reference = 'IM8888.CCFX' # 主力合约 g.holding_long = False # 持仓状态 g.entry_price = 0.0 # 入场价格 # 策略参数 g.params = { 'profit_target': 0.03, # 3%止盈 'stop_loss': 0.01, # 1%止损 } # 基本设置 set_benchmark(g.benchmark) set_option('use_real_price', True) set_option("avoid_future_data", True) log.set_level('order', 'error') # 期货账户设置 set_subportfolios([SubPortfolioConfig(cash=context.portfolio.starting_cash, type='index_futures')]) set_order_cost(OrderCost(open_commission=0.000023, close_commission=0.000023, close_today_commission=0.000023), type='index_futures') set_option('futures_margin_rate', 0.15) set_slippage(StepRelatedSlippage(2)) # 运行计划 run_daily(trade, time='every_bar', reference_security=g.reference) def trade(context): # 获取当前主力合约 contracts = [c for c in get_all_securities(['futures']).index if c.startswith(g.future_type)] if not contracts: return contract = sorted(contracts, key=lambda x: get_security_info(x).end_date)[0] # 获取行情数据 current_data = get_current_data() current_price = current_data[contract].last_price time_str = context.current_dt.strftime('%H%M') # 过滤无效时间和涨跌停 if not ('0930' <= time_str <= '1455') or not if_order(contract): return # 获取最近120根1分钟K线 bars = get_bars(contract, count=60, unit='1m', fields=['close', 'high', 'low', 'volume'], include_now=True, df=True) if len(bars) < 20: return closes = bars['close'].values highs = bars['high'].values lows = bars['low'].values # 计算核心指标 ma5 = talib.EMA(closes, 5)[-1] # 5分钟均线 ma10 = talib.EMA(closes, 10)[-1] # 10分钟均线 rsi = talib.RSI(closes, 7)[-1] # 7周期RSI # === 简单粗暴入场条件 === if not g.holding_long: # 条件1: 价格突破前高 price_breakout = current_price > np.max(highs[-10:-1]) # 条件2: 均线多头排列 (MA5 > MA10) ma_bullish = ma5 > ma10 # 条件3: RSI超卖 (低于30) rsi_oversold = rsi < 30 # 满足任意两个条件即入场 if (price_breakout and ma_bullish) or (price_breakout and rsi_oversold) or (ma_bullish and rsi_oversold): order(contract, 1, side='long') g.holding_long = True g.entry_price = current_price log.info(f"{time_str} 开多 @ {current_price:.2f}") # === 持仓处理 === if g.holding_long: profit_ratio = (current_price - g.entry_price) / g.entry_price # 止盈条件：达到3%盈利 if profit_ratio >= g.params['profit_target']: order_target(contract, 0, side='long') g.holding_long = False log.info(f"止盈平仓 @ {current_price:.2f}, 盈利:{profit_ratio100:.2f}%") return # 止损条件：亏损1% if profit_ratio <= -g.params['stop_loss']: order_target(contract, 0, side='long') g.holding_long = False log.info(f"止损平仓 @ {current_price:.2f}, 亏损:{abs(profit_ratio)100:.2f}%") return # 尾盘无条件平仓 (14:55) if time_str >= '1455': order_target(contract, 0, side='long') g.holding_long = False profit = (current_price - g.entry_price) / g.entry_price * 100 log.info(f"尾盘平仓 @ {current_price:.2f}, 盈亏:{profit:.2f}%") # 判断是否可以下单（非涨跌停状态） def if_order(security): current_data = get_current_data() if security not in current_data: return False high_limit = current_data[security].high_limit last_price = current_data[security].last_price low_limit = current_data[security].low_limit return last_price != high_limit and last_price != low_limit def after_market_close(context): log.info('当日交易结束') 详细解析这段策略并画出流程图

1. **initialize(context)**: 初始化函数，设置全局变量、参数、基准等。 - 交易品种：IM期货 - 基准：中证1000指数 - 参考主力合约：IM8888.CCFX - 持仓状态：初始为False - 入场价格：0.0 - 参数：止盈3%，...

from jqdata import * from jqfactor import get_factor_values import datetime import math from scipy.optimize import minimize import pandas as pd # 初始化函数，设定基准等等 def initialize(context): # 设定沪深300作为基准 set_benchmark("399303.XSHE") # 打开防未来函数 set_option("avoid_future_data", True) # 开启动态复权模式(真实价格) set_option("use_real_price", True) # 输出内容到日志 log.info() log.info("初始函数开始运行") # 过滤掉order系列API产生的比error级别低的log log.set_level("order", "error") # 固定滑点设置ETF 0.001(即交易对手方一档价) set_slippage(FixedSlippage(0.002), type="fund") # 股票交易总成本0.3%(含固定滑点0.02) set_order_cost( OrderCost( open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003, close_today_commission=0, min_commission=5, ), type="stock", ) g.hold_list = [] # 记录策略的持仓股票 g.positions = {} # 记录策略的持仓股票 # 持仓股票数 g.stock_sum = 6 # 判断买卖点的行业数量 g.num = 1 # 空仓的月份 g.pass_months = [] # 策略执行计划 run_weekly(adjust, 1, "9:31") run_daily(check, "14:50") # 获取昨日涨停票并卖出 def check(context): # 获取已持有列表 g.hold_list = list(g.positions.keys()) banner_stocks = [] # 获取昨日涨停列表 if g.hold_list != []: df = get_price( g.hold_list, end_date=context.previous_date, frequency="daily", fields=["close", "high_limit"], count=1, panel=False, fill_paused=False, ) df = df[df["close"] == df["high_limit"]] banner_stocks = list(df.code) for stock in banner_stocks: order_target_value_(context, stock, 0) # 获取昨日跌停列表 if g.hold_list != []: df = get_price( g.hold_list, end_date=context.previous_date, frequency="daily", fields=["close", "low_limit"], count=1, panel=False, fill_paused=False, ) df = df[df["close"] == df["low_limit"]] banner_stocks = list(df.code) for stock in banner_stocks: order_target_value_(context, stock, 0) # 获取策略当前持仓市值 def get_total_value(context): return sum(context.portfolio.positions[key].price * value for key, value in g.positions.items()) # 调仓 def adjust(context): target = select(context) # 获取前stock_sum个标的 target = target[:min(len(target), g.stock_sum)] # 获取已持有列表 g.hold_list = list(g.positions.keys()) portfolio = context.portfolio # 调仓卖出 for stock in g.hold_list: if stock not in target: order_target_value_(context, stock, 0) # 调仓买入 count = len(set(target) - set(g.hold_list)) if count == 0: return # 目标市值 target_value = portfolio.total_value # 当前市值 position_value = get_total_value(context) # 可用现金:当前现金 available_cash = portfolio.available_cash # 买入股票的总市值 value = max(0, min(target_value - position_value, available_cash)) # 等价值买入每一个未买入的标的 for security in target: if security not in g.hold_list: order_target_value_(context, security, value / count) # 择时 def select(context): I = get_market_breadth(context) industries = {"银行I", "煤炭I", "采掘I", "钢铁I"} if not industries.intersection(I) and not is_empty_month(context): return filter(context) return [] # 获取市场 def get_market_breadth(context): # 指定日期防止未来数据 yesterday = context.previous_date # 获取初始列表中证全指（000985.XSHG） stocks = get_index_stocks("000985.XSHG") count = 1 h = get_price( stocks, end_date=yesterday, frequency="1d", fields=["close"], count=count + 20, panel=False, ) h["date"] = pd.DatetimeIndex(h.time).date # 将长表格转换为宽表格，方便按日期分析股票价格。 df_close = h.pivot(index="code", columns="date", values="close").dropna(axis=0) # 计算20日均线 df_ma20 = df_close.rolling(window=20, axis=1).mean().iloc[:, -count:] # 计算偏离程度 df_bias = df_close.iloc[:, -count:] > df_ma20 df_bias["industry_name"] = getStockIndustry(stocks) # 计算行业偏离比例 df_ratio = ((df_bias.groupby("industry_name").sum() * 100.0) / df_bias.groupby("industry_name").count()).round() # 获取偏离程度最高的行业 top_values = df_ratio.loc[:, yesterday].nlargest(g.num) I = top_values.index.tolist() return I # 基础过滤(过滤科创北交、ST、停牌、次新股) def filter_basic_stock(context, stock_list): # 30开头的是深交所的创业板， # 68开头的是上交所的科创板， # 8开头的股票可能指的是北交所的， # 新三板北交所的股票代码通常以43、83、87等开头 # 4开头的股票可能属于退市板块 current_data = get_current_data() return [ stock for stock in stock_list if not current_data[stock].paused and not current_data[stock].is_st and "ST" not in current_data[stock].name and "" not in current_data[stock].name and "退" not in current_data[stock].name and not (stock[0] == "4" or stock[0] == "8" or stock[:2] == "68") and not context.previous_date - get_security_info(stock).start_date < datetime.timedelta(375) ] # 过滤当前时间涨跌停的股票 def filter_limitup_limitdown_stock(stock_list): current_data = get_current_data() return [ stock for stock in stock_list if current_data[stock].last_price < current_data[stock].high_limit and current_data[stock].last_price > current_data[stock].low_limit ] # 判断今天是在空仓月 def is_empty_month(context): month = context.current_dt.month return month in g.pass_months def getStockIndustry(stocks): # 第一步：获取原始行业数据（假设stocks是股票代码列表） industry = get_industry(stocks) # 第二步：提取申万一级行业名称 return pd.Series({stock: info["sw_l1"]["industry_name"] for stock, info in industry.items() if "sw_l1" in info}) # 过滤股票 def filter(context): stocks = get_index_stocks("399303.XSHE") # 这里的有问题，需要由399303.XSHE代替 stocks = filter_basic_stock(context, stocks) stocks = ( get_fundamentals( query( valuation.code, ) .filter( valuation.code.in_(stocks), # 从现有股票池中筛选 indicator.adjusted_profit > 0, # 要求调整后净利润>0 ) .order_by(valuation.market_cap.asc()) # 按市值升序排列（从小市值开始） ) .head(20) # 取前20只股票 .code # 提取股票代码 ) stocks = filter_limitup_limitdown_stock(stocks) return stocks # 自定义下单(涨跌停不交易) def order_target_value_(context, security, value): current_data = get_current_data() # 检查标的是否停牌、涨停、跌停 if current_data[security].paused: log.info(f"{security}: 今日停牌") return False # 检查是否涨停 if current_data[security].last_price == current_data[security].high_limit: log.info(f"{security}: 当前涨停") return False # 检查是否跌停 if current_data[security].last_price == current_data[security].low_limit: log.info(f"{security}: 当前跌停") return False # 获取当前标的的价格 price = current_data[security].last_price # 获取当前策略的持仓数量 current_position = g.positions.get(security, 0) # 计算目标持仓数量 target_position = (int(value / price) // 100) 100 if price != 0 else 0 # 计算需要调整的数量 adjustment = target_position - current_position # 检查是否当天买入卖出 closeable_amount = context.portfolio.positions[security].closeable_amount if security in context.portfolio.positions else 0 if adjustment < 0 and closeable_amount == 0: log.info(f"{security}: 当天买入不可卖出") return False # 下单并更新持仓 if adjustment != 0: o = order(security, adjustment) if o: # 更新持仓数量 amount = o.amount if o.is_buy else -o.amount g.positions[security] = amount + current_position # 如果目标持仓为零，移除该证券 if target_position == 0: g.positions.pop(security, None) # 更新持有列表 g.hold_list = list(g.positions.keys()) return True return False （把这个策略的代码格式转换成backtrade）

聚宽的initialize函数对应Backtrader的__init__方法，用于初始化参数；handle_data函数对应next方法，用于处理每个交易日的逻辑。基于引用[2]的布林带策略： python class BollingerStrategy(bt....

# -- coding: utf-8 -- import sys import os import cv2 import numpy as np import time from PyQt5.QtWidgets import ( QApplication, QMainWindow, QPushButton, QWidget, QVBoxLayout, QHBoxLayout, QMessageBox, QLabel, QFileDialog, QToolBar, QComboBox, QStatusBar, QGroupBox, QSlider, QDockWidget, QProgressDialog, QLineEdit, QRadioButton, QGridLayout, QSpinBox ) from PyQt5.QtCore import QRect, Qt, QSettings, QThread, pyqtSignal #sys.path.append("D:\\海康\\MVS\\Development\\Samples\\Python\\wanzheng.py") import ctypes from ctypes import cast, POINTER from datetime import datetime import logging import socket import serial import skimage import platform from mv_import.Cam_eraConstants import * import threading import time import types # 在导入Cam_Operation_class之前添加 import importlib # clothing_inspection.py # 强制重载关键模块 def reload_camera_modules(): modules_to_reload = [ 'MvImport.CamOperation_class', 'MvImport.MvCameraControl_class', 'MvImport.Camera_Params_header', 'MvImport.MvErrorDefine_const' ] for module_name in modules_to_reload: if module_name in sys.modules: print(f"强制重载模块: {module_name}") module = sys.modules[module_name] importlib.reload(module) # 清除缓存 for key in list(sys.modules.keys()): if key.startswith('MvImport'): del sys.modules[key] print("模块重载完成") # 在导入前调用 reload_camera_modules() from mv_import.Cam_Operation_class import CameraOperation import mv_import.Camera_Params_const # 设置Python包环境变量 os.environ["PACKAGE_CONTEXT"] = "main" # 标识当前在包环境中 # 获取当前脚本所在目录 (main目录) current_dir = os.path.dirname(os.path.abspath(file)) # 添加main目录到Python路径 if current_dir not in sys.path: sys.path.insert(0, current_dir) # 添加MvImport包路径 mv_import_path = os.path.join(current_dir, "MvImport") if mv_import_path not in sys.path: sys.path.append(mv_import_path) # 强制设置包上下文 import MvImport MvImport.package = "MvImport" # 显式设置包名 # 程序主要逻辑... # 获取当前脚本所在目录 (main目录) current_dir = os.path.dirname(os.path.abspath(file)) # 添加main目录到Python路径 if current_dir not in sys.path: sys.path.insert(0, current_dir) # 添加MvImport包路径 mv_import_path = os.path.join(current_dir, "MvImport") if mv_import_path not in sys.path: sys.path.append(mv_import_path) # 加载必要的DLL - 根据实际路径调整 dll_path = os.path.join(current_dir, "..", "dlls", "MvCameraControl.dll") try: ctypes.CDLL(dll_path) except OSError as e: print(f"无法加载相机SDK: {e}") sys.exit(1) class PathFixer: """自动修复模块导入路径的工具类""" def init(self): self.original_path = sys.path.copy() self.project_root = self.find_project_root() self.add_project_path() def find_project_root(self): """自动定位项目根目录""" # 获取当前脚本所在目录 current_dir = os.path.dirname(os.path.abspath(file)) # 向上查找包含 .git 或 requirements.txt 的目录 root = current_dir while not self.is_project_root(root) and root != os.path.dirname(root): root = os.path.dirname(root) return root def is_project_root(self, path): """检查目录是否包含项目标志文件""" return ( os.path.exists(os.path.join(path, ".git")) or os.path.exists(os.path.join(path, "requirements.txt")) or os.path.exists(os.path.join(path, "README.md")) ) def add_project_path(self): """添加项目路径到sys.path""" if self.project_root not in sys.path: sys.path.insert(0, self.project_root) print(f"已添加项目根目录到路径: {self.project_root}") def restore_path(self): """恢复原始路径设置""" sys.path = self.original_path def setup_path(): """确保项目根目录在模块搜索路径中""" # 获取当前脚本所在目录 script_dir = os.path.dirname(os.path.abspath(file)) # 添加项目根目录到模块搜索路径 if script_dir not in sys.path: sys.path.insert(0, script_dir) print(f"已添加路径到 sys.path: {script_dir}") # 调试信息 print("\n当前工作目录:", os.getcwd()) print("Python 模块搜索路径:") for path in sys.path: print(f" - {path}") # 在导入其他模块前设置路径 setup_path() class ManagedThread(threading.Thread): def init(self, *args, **kwargs): super().init(*args, **kwargs) self._stop_event = threading.Event() # 设置为非守护线程 self.daemon = False def stop(self): """安全停止线程""" self._stop_event.set() def should_stop(self): """检查是否应该停止""" return self._stop_event.is_set() def worker(): """线程工作函数""" try: while not threading.current_thread().should_stop(): # 模拟工作 time.sleep(1) # 安全输出 sys.stdout.write("Working...\n") except Exception as e: # 避免在关闭时使用stderr pass def main(): # 创建并启动线程 threads = [] for _ in range(3): t = ManagedThread(target=worker) t.start() threads.append(t) try: # 主程序逻辑 time.sleep(5) finally: # 安全停止所有线程 for t in threads: t.stop() for t in threads: t.join(timeout=2.0) # 设置超时避免无限等待 # 确保所有输出完成 sys.stdout.flush() sys.stderr.flush() # 在导入部分添加 from MvImport.CameraParams_header import ( MV_GIGE_DEVICE, MV_USB_DEVICE, MV_GENTL_CAMERALINK_DEVICE, MV_GENTL_CXP_DEVICE, MV_GENTL_XOF_DEVICE ) # 获取当前文件所在目录 current_dir = os.path.dirname(os.path.abspath(file)) # ===== 路径修复 ===== sdk_path = os.path.join(current_dir, "MvImport") if sdk_path not in sys.path: sys.path.append(sdk_path) def fix_sdk_path(): """修复海康SDK的加载路径""" if getattr(sys, 'frozen', False): # 打包模式 base_path = sys._MEIPASS # 添加DLL目录到系统路径 dll_path = os.path.join(base_path, "dlls") os.environ['PATH'] = dll_path + os.pathsep + os.environ['PATH'] try: # 直接加载DLL ctypes.WinDLL(os.path.join(dll_path, "MvCamCtrldll.dll")) ctypes.WinDLL(os.path.join(dll_path, "MvCameraControl.dll")) except OSError as e: logging.error(f"核心DLL加载失败: {e}") sys.exit(1) else: # 开发模式 # 确保SDK路径存在 if sdk_path not in sys.path: sys.path.append(sdk_path) # 添加DLL到系统路径 dll_dir = r"D:\海康\MVS\Runtime\Win64" if dll_dir not in os.environ['PATH']: os.environ['PATH'] = dll_dir + os.pathsep + os.environ['PATH'] # 立即执行路径修复 fix_sdk_path() # ===== 正确导入SDK模块 ===== try: from MvImport.MvCameraControl_class import MvCamera print("成功导入MvCamera类") from MvImport.CameraParams_header import * from MvImport.MvErrorDefine_const import * except ImportError as e: print(f"SDK导入失败: {e}") sys.exit(1) # 配置日志系统 logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("cloth_inspection_debug.log"), logging.StreamHandler() ] ) logging.info("布料印花检测系统启动") # 全局变量 current_sample_path = "" detection_history = [] isGrabbing = False isOpen = False obj_cam_operation = None frame_monitor_thread = None sensor_monitor_thread = None sensor_controller = None MV_OK = 0 MV_E_CALLORDER = -2147483647 # ==================== 传感器通讯模块 ==================== class SensorController: def init(self): self.connected = False self.running = False self.connection = None def connect(self, config): try: if config['type'] == 'serial': self.connection = serial.Serial( port=config['port'], baudrate=config['baudrate'], timeout=config.get('timeout', 1.0) ) else: self.connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.connection.connect((config['ip'], config['port'])) self.connection.settimeout(config.get('timeout', 1.0)) self.connected = True self.running = True logging.info(f"传感器连接成功: {config}") return True except Exception as e: logging.error(f"传感器连接失败: {str(e)}") return False def disconnect(self): if self.connection: try: self.connection.close() except: pass self.connection = None self.connected = False self.running = False logging.info("传感器已断开") def read_data(self): if not self.connected: return None return { 'tension': np.random.uniform(10.0, 20.0), 'speed': np.random.uniform(1.0, 5.0), 'temperature': np.random.uniform(20.0, 30.0), 'humidity': np.random.uniform(40.0, 60.0) } def wait_for_material(self, delay_seconds=0): if not self.connected: logging.warning("未连接传感器，跳过等待") return False logging.info(f"等待布料到达，延迟 {delay_seconds} 秒") start_time = time.time() while time.time() - start_time < delay_seconds: QThread.msleep(100) if not self.running: return False logging.info("布料已到位，准备拍摄") return True class SensorMonitorThread(QThread): data_updated = pyqtSignal(dict) def init(self, sensor_controller): super().init() self.sensor_controller = sensor_controller self.running = True def run(self): while self.running: if self.sensor_controller and self.sensor_controller.connected: try: data = self.sensor_controller.read_data() if data: self.data_updated.emit(data) except Exception as e: logging.error(f"传感器数据读取错误: {str(e)}") QThread.msleep(500) def stop(self): self.running = False self.wait(2000) def wait_for_material(self, delay_seconds): return self.sensor_controller.wait_for_material(delay_seconds) # ==================== 相机帧监控线程 ==================== class FrameMonitorThread(QThread): frame_status = pyqtSignal(str) # 用于发送状态消息的信号 def init(self, cam_operation): super().init() self.cam_operation = cam_operation self.running = True self.frame_count = 0 self.last_time = time.time() def run(self): """监控相机帧状态的主循环""" while self.running: try: if self.cam_operation and self.cam_operation.is_grabbing: # 获取帧统计信息 frame_info = self.get_frame_info() if frame_info: fps = frame_info.get('fps', 0) dropped = frame_info.get('dropped', 0) status = f"FPS: {fps:.1f} | 丢帧: {dropped}" self.frame_status.emit(status) else: self.frame_status.emit("取流中...") else: self.frame_status.emit("相机未取流") except Exception as e: self.frame_status.emit(f"监控错误: {str(e)}") # 每500ms检查一次 QThread.msleep(500) def stop(self): """停止监控线程""" self.running = False self.wait(1000) # 等待线程结束 def calculate_fps(self): """计算当前帧率""" current_time = time.time() elapsed = current_time - self.last_time if elapsed > 0: fps = self.frame_count / elapsed self.frame_count = 0 self.last_time = current_time return fps return 0 def get_frame_info(self): """获取帧信息""" try: # 更新帧计数 self.frame_count += 1 # 返回帧信息 return { 'fps': self.calculate_fps(), 'dropped': 0 # 实际应用中需要从相机获取真实丢帧数 } except Exception as e: logging.error(f"获取帧信息失败: {str(e)}") return None # ==================== 优化后的检测算法 ==================== def enhanced_check_print_quality(sample_image_path, test_image, threshold=0.05, sensor_data=None): if sensor_data: speed_factor = min(1.0 + sensor_data['speed'] * 0.1, 1.5) env_factor = 1.0 + abs(sensor_data['temperature'] - 25) * 0.01 + abs(sensor_data['humidity'] - 50) * 0.005 adjusted_threshold = threshold * speed_factor * env_factor logging.info(f"根据传感器数据调整阈值: 原始={threshold:.4f}, 调整后={adjusted_threshold:.4f}") else: adjusted_threshold = threshold try: sample_img_data = np.fromfile(sample_image_path, dtype=np.uint8) sample_image = cv2.imdecode(sample_img_data, cv2.IMREAD_GRAYSCALE) if sample_image is None: logging.error(f"无法解码样本图像: {sample_image_path}") return None, None, None except Exception as e: logging.exception(f"样本图像读取异常: {str(e)}") return None, None, None if len(test_image.shape) == 3: test_image_gray = cv2.cvtColor(test_image, cv2.COLOR_BGR2GRAY) else: test_image_gray = test_image.copy() sample_image = cv2.GaussianBlur(sample_image, (5, 5), 0) test_image_gray = cv2.GaussianBlur(test_image_gray, (5, 5), 0) try: orb = cv2.ORB_create(nfeatures=200) keypoints1, descriptors1 = orb.detectAndCompute(sample_image, None) keypoints2, descriptors2 = orb.detectAndCompute(test_image_gray, None) if descriptors1 is None or descriptors2 is None: logging.warning("无法提取特征描述符，跳过配准") aligned_sample = sample_image else: bf = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=True) matches = bf.match(descriptors1, descriptors2) matches = sorted(matches, key=lambda x: x.distance) if len(matches) > 10: src_pts = np.float32([keypoints1[m.queryIdx].pt for m in matches]).reshape(-1, 1, 2) dst_pts = np.float32([keypoints2[m.trainIdx].pt for m in matches]).reshape(-1, 1, 2) H, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0) if H is not None: aligned_sample = cv2.warpPerspective( sample_image, H, (test_image_gray.shape[1], test_image_gray.shape[0]) ) logging.info("图像配准成功，使用配准后样本") else: aligned_sample = sample_image logging.warning("无法计算单应性矩阵，使用原始样本") else: aligned_sample = sample_image logging.warning("特征点匹配不足，跳过图像配准") except Exception as e: logging.error(f"图像配准失败: {str(e)}") aligned_sample = sample_image try: if aligned_sample.shape != test_image_gray.shape: test_image_gray = cv2.resize(test_image_gray, (aligned_sample.shape[1], aligned_sample.shape[0])) except Exception as e: logging.error(f"图像调整大小失败: {str(e)}") return None, None, None try: from skimage.metrics import structural_similarity as compare_ssim ssim_score, ssim_diff = compare_ssim( aligned_sample, test_image_gray, full=True, gaussian_weights=True, data_range=255 ) except ImportError: from skimage.measure import compare_ssim ssim_score, ssim_diff = compare_ssim( aligned_sample, test_image_gray, full=True, gaussian_weights=True ) except Exception as e: logging.error(f"SSIM计算失败: {str(e)}") abs_diff = cv2.absdiff(aligned_sample, test_image_gray) ssim_diff = abs_diff.astype(np.float32) / 255.0 ssim_score = 1.0 - np.mean(ssim_diff) ssim_diff = (1 - ssim_diff) * 255 abs_diff = cv2.absdiff(aligned_sample, test_image_gray) combined_diff = cv2.addWeighted(ssim_diff.astype(np.uint8), 0.7, abs_diff, 0.3, 0) _, thresholded = cv2.threshold(combined_diff, 30, 255, cv2.THRESH_BINARY) kernel = np.ones((3, 3), np.uint8) thresholded = cv2.morphologyEx(thresholded, cv2.MORPH_OPEN, kernel) thresholded = cv2.morphologyEx(thresholded, cv2.MORPH_CLOSE, kernel) diff_pixels = np.count_nonzero(thresholded) total_pixels = aligned_sample.size diff_ratio = diff_pixels / total_pixels is_qualified = diff_ratio <= adjusted_threshold marked_image = cv2.cvtColor(test_image_gray, cv2.COLOR_GRAY2BGR) marked_image[thresholded == 255] = [0, 0, 255] labels = skimage.measure.label(thresholded) properties = skimage.measure.regionprops(labels) for prop in properties: if prop.area > 50: y, x = prop.centroid cv2.putText(marked_image, f"Defect", (int(x), int(y)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 255), 1) return is_qualified, diff_ratio, marked_image # ==================== 传感器控制的质量检测流程 ==================== def sensor_controlled_check(): global isGrabbing, obj_cam_operation, current_sample_path, detection_history, sensor_controller logging.info("质量检测启动") sensor_data = None if sensor_controller and sensor_controller.connected: sensor_data = sensor_controller.read_data() if not sensor_data: QMessageBox.warning(mainWindow, "传感器警告", "无法读取传感器数据，将使用默认参数", QMessageBox.Ok) else: logging.info("未连接传感器，使用默认参数检测") check_print_with_sensor(sensor_data) def check_print_with_sensor(sensor_data=None): global isGrabbing, obj_cam_operation, current_sample_path, detection_history logging.info("检测印花质量按钮按下") if not isGrabbing: QMessageBox.warning(mainWindow, "错误", "请先开始取流并捕获图像！", QMessageBox.Ok) return if not obj_cam_operation: QMessageBox.warning(mainWindow, "错误", "相机未正确初始化！", QMessageBox.Ok) return if not current_sample_path or not os.path.exists(current_sample_path): QMessageBox.warning(mainWindow, "错误", "请先设置有效的标准样本图像！", QMessageBox.Ok) return progress = QProgressDialog("正在检测...", "取消", 0, 100, mainWindow) progress.setWindowModality(Qt.WindowModal) progress.setValue(10) try: test_image = obj_cam_operation.get_current_frame() progress.setValue(30) if test_image is None: QMessageBox.warning(mainWindow, "错误", "无法获取当前帧图像！", QMessageBox.Ok) return diff_threshold = mainWindow.sliderDiffThreshold.value() / 100.0 logging.info(f"使用差异度阈值: {diff_threshold}") progress.setValue(50) is_qualified, diff_ratio, marked_image = enhanced_check_print_quality( current_sample_path, test_image, threshold=diff_threshold, sensor_data=sensor_data ) progress.setValue(70) if is_qualified is None: QMessageBox.critical(mainWindow, "检测错误", "检测失败，请检查日志", QMessageBox.Ok) return logging.info(f"检测结果: 合格={is_qualified}, 差异={diff_ratio}") progress.setValue(90) update_diff_display(diff_ratio, is_qualified) result_text = f"印花是否合格: {'合格' if is_qualified else '不合格'}\n差异占比: {diff_ratio100:.2f}%\n阈值: {diff_threshold100:.2f}%" QMessageBox.information(mainWindow, "检测结果", result_text, QMessageBox.Ok) if marked_image is not None: cv2.imshow("缺陷标记结果", marked_image) cv2.waitKey(0) cv2.destroyAllWindows() detection_result = { 'timestamp': datetime.now(), 'qualified': is_qualified, 'diff_ratio': diff_ratio, 'threshold': diff_threshold, 'sensor_data': sensor_data if sensor_data else {} } detection_history.append(detection_result) update_history_display() progress.setValue(100) except Exception as e: logging.exception("印花检测失败") QMessageBox.critical(mainWindow, "检测错误", f"检测过程中发生错误: {str(e)}", QMessageBox.Ok) finally: progress.close() def update_diff_display(diff_ratio, is_qualified): mainWindow.lblCurrentDiff.setText(f"当前差异度: {diff_ratio100:.2f}%") if is_qualified: mainWindow.lblDiffStatus.setText("状态: 合格") mainWindow.lblDiffStatus.setStyleSheet("color: green; font-size: 12px;") else: mainWindow.lblDiffStatus.setText("状态: 不合格") mainWindow.lblDiffStatus.setStyleSheet("color: red; font-size: 12px;") def update_diff_threshold(value): mainWindow.lblDiffValue.setText(f"{value}%") def save_sample_image(): global isGrabbing, obj_cam_operation, current_sample_path if not isGrabbing: QMessageBox.warning(mainWindow, "错误", "请先开始取流并捕获图像！", QMessageBox.Ok) return # 检查是否有可用帧 if not obj_cam_operation.is_frame_available(): QMessageBox.warning(mainWindow, "无有效图像", "未捕获到有效图像，请检查相机状态！", QMessageBox.Ok) return settings = QSettings("ClothInspection", "CameraApp") last_dir = settings.value("last_save_dir", os.path.join(os.getcwd(), "captures")) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_filename = f"sample_{timestamp}" file_path, selected_filter = QFileDialog.getSaveFileName( mainWindow, "保存标准样本图像", os.path.join(last_dir, default_filename), "BMP Files (.bmp);;PNG Files (.png);;JPEG Files (.jpg);;所有文件 ()", options=QFileDialog.DontUseNativeDialog ) if not file_path: return file_extension = os.path.splitext(file_path)[1].lower() if not file_extension: if "BMP" in selected_filter: file_path += ".bmp" elif "PNG" in selected_filter: file_path += ".png" elif "JPEG" in selected_filter or "JPG" in selected_filter: file_path += ".jpg" else: file_path += ".bmp" file_extension = os.path.splitext(file_path)[1].lower() format_mapping = {".bmp": "bmp", ".png": "png", ".jpg": "jpg", ".jpeg": "jpg"} save_format = format_mapping.get(file_extension) if not save_format: QMessageBox.warning(mainWindow, "错误", "不支持的文件格式！", QMessageBox.Ok) return directory = os.path.dirname(file_path) if directory and not os.path.exists(directory): try: os.makedirs(directory, exist_ok=True) except OSError as e: QMessageBox.critical(mainWindow, "目录创建错误", f"无法创建目录 {directory}: {str(e)}", QMessageBox.Ok) return try: ret = obj_cam_operation.save_image(file_path, save_format) if ret != MV_OK: strError = f"保存样本图像失败: {hex(ret)}" QMessageBox.warning(mainWindow, "错误", strError, QMessageBox.Ok) else: QMessageBox.information(mainWindow, "成功", f"标准样本已保存至:\n{file_path}", QMessageBox.Ok) current_sample_path = file_path update_sample_display() settings.setValue("last_save_dir", os.path.dirname(file_path)) except Exception as e: QMessageBox.critical(mainWindow, "异常错误", f"保存图像时发生错误: {str(e)}", QMessageBox.Ok) def preview_sample(): global current_sample_path if not current_sample_path or not os.path.exists(current_sample_path): QMessageBox.warning(mainWindow, "错误", "请先设置有效的标准样本图像！", QMessageBox.Ok) return try: img_data = np.fromfile(current_sample_path, dtype=np.uint8) sample_img = cv2.imdecode(img_data, cv2.IMREAD_COLOR) if sample_img is None: raise Exception("无法加载图像") cv2.imshow("标准样本预览", sample_img) cv2.waitKey(0) cv2.destroyAllWindows() except Exception as e: QMessageBox.warning(mainWindow, "错误", f"预览样本失败: {str(e)}", QMessageBox.Ok) def update_sample_display(): global current_sample_path if current_sample_path: mainWindow.lblSamplePath.setText(f"当前样本: {os.path.basename(current_sample_path)}") mainWindow.lblSamplePath.setToolTip(current_sample_path) mainWindow.bnPreviewSample.setEnabled(True) else: mainWindow.lblSamplePath.setText("当前样本: 未设置样本") mainWindow.bnPreviewSample.setEnabled(False) def update_history_display(): global detection_history mainWindow.cbHistory.clear() for i, result in enumerate(detection_history[-10:]): timestamp = result['timestamp'].strftime("%H:%M:%S") status = "合格" if result['qualified'] else "不合格" ratio = f"{result['diff_ratio']100:.2f}%" mainWindow.cbHistory.addItem(f"[极客{timestamp}] {status} - 差异: {ratio}") def TxtWrapBy(start_str, end, all): start = all.find(start_str) if start >= 0: start += len(start_str) end = all.find(end, start) if end >= 0: return all[start:end].strip() def ToHexStr(num): if not isinstance(num, int): try: num = int(num) except: return f"<非整数:{type(num)}>" chaDic = {10: 'a', 11: 'b', 12: 'c', 13: 'd', 14: 'e', 15: 'f'} hexStr = "" if num < 0: num = num + 2 ** 32 while num >= 16: digit = num % 16 hexStr = chaDic.get(digit, str(digit)) + hexStr num //= 16 hexStr = chaDic.get(num, str(num)) + hexStr return "0x" + hexStr def decoding_char(c_ubyte_value): c_char_p_value = ctypes.cast(c_ubyte_value, ctypes.c_char_p) try: decode_str = c_char_p_value.value.decode('gbk') except UnicodeDecodeError: decode_str = str(c_char_p_value.value) return decode_str def enum_devices(): global deviceList, obj_cam_operation n_layer_type = ( MV_GIGE_DEVICE | MV_USB_DEVICE | MV_GENTL_CAMERALINK_DEVICE | MV_GENTL_CXP_DEVICE | MV_GENTL_XOF_DEVICE ) # 创建设备列表 deviceList = MV_CC_DEVICE_INFO_LIST() # 枚举设备 ret = MvCamera.MV_CC_EnumDevices(n_layer_type, deviceList) if ret != MV_OK: error_msg = f"枚举设备失败! 错误码: 0x{ret:x}" logging.error(error_msg) QMessageBox.warning(mainWindow, "错误", error_msg, QMessageBox.Ok) return ret if deviceList.nDeviceNum == 0: QMessageBox.warning(mainWindow, "提示", "未找到任何设备", QMessageBox.Ok) return MV_OK logging.info(f"找到 {deviceList.nDeviceNum} 个设备") # 处理设备信息 devList = [] for i in range(deviceList.nDeviceNum): # 获取设备信息 mvcc_dev_info = ctypes.cast( deviceList.pDeviceInfo[i], ctypes.POINTER(MV_CC_DEVICE_INFO) ).contents # 根据设备类型提取信息 if mvcc_dev_info.nTLayerType == MV_GIGE_DEVICE: st_gige_info = mvcc_dev_info.SpecialInfo.stGigEInfo ip_addr = ( f"{(st_gige_info.nCurrentIp >> 24) & 0xFF}." f"{(st_gige_info.nCurrentIp >> 16) & 0xFF}." f"{(st_gige_info.nCurrentIp >> 8) & 0xFF}." f"{st_gige_info.nCurrentIp & 0xFF}" ) # 修复：将c_ubyte_Array_16转换为字节串再解码 user_defined_bytes = bytes(st_gige_info.chUserDefinedName) dev_name = f"GigE: {user_defined_bytes.decode('gbk', 'ignore')}" devList.append(f"[{i}] {dev_name} ({ip_addr})") elif mvcc_dev_info.nTLayerType == MV_USB_DEVICE: st_usb_info = mvcc_dev_info.SpecialInfo.stUsb3VInfo serial = bytes(st_usb_info.chSerialNumber).decode('ascii', 'ignore').rstrip('\x00') # 修复：同样处理用户自定义名称 user_defined_bytes = bytes(st_usb_info.chUserDefinedName) dev_name = f"USB: {user_defined_bytes.decode('gbk', 'ignore')}" devList.append(f"[{i}] {dev_name} (SN: {serial})") else: devList.append(f"[{i}] 未知设备类型: {mvcc_dev_info.nTLayerType}") # 更新UI mainWindow.ComboDevices.clear() mainWindow.ComboDevices.addItems(devList) if devList: mainWindow.ComboDevices.setCurrentIndex(0) mainWindow.statusBar().showMessage(f"找到 {deviceList.nDeviceNum} 个设备", 3000) return MV_OK # clothing_inspection.py def initialize_camera(): from MvImport.CamOperation_class import CameraOperation # 创建相机操作实例 cam_operator = CameraOperation() # 打开设备 - 添加错误处理 try: if cam_operator.open_device(): print("相机设备已成功打开") else: print("无法打开相机设备") except AttributeError as e: print(f"方法调用错误: {e}") # 检查类实现 if not hasattr(CameraOperation, 'open_device'): print("错误: CameraOperation 类缺少 open_device 方法") # 动态添加方法作为临时解决方案 from types import MethodType def temp_open_device(self): print("警告: 使用临时open_device实现") return False CameraOperation.open_device = MethodType(temp_open_device, CameraOperation) cam_operator.open_device() # ===== 关键改进：相机操作函数 ===== def open_device(): global deviceList, nSelCamIndex, obj_cam_operation, isOpen, frame_monitor_thread, mainWindow if isOpen: QMessageBox.warning(mainWindow, "Error", '相机已打开!', QMessageBox.Ok) return MV_E_CALLORDER nSelCamIndex = mainWindow.ComboDevices.currentIndex() if nSelCamIndex < 0: QMessageBox.warning(mainWindow, "Error", '请选择相机!', QMessageBox.Ok) return MV_E_CALLORDER # 创建相机控制对象 cam = MvCamera() # 初始化相机操作对象 - 确保传入有效的相机对象 obj_cam_operation = CameraOperation(cam, deviceList, nSelCamIndex) # 调试信息 logging.debug(f"CameraOperation类方法: {dir(CameraOperation)}") logging.debug(f"obj_cam_operation实例方法: {dir(obj_cam_operation)}") # 如果实例缺少方法，但类有该方法，手动绑定 if hasattr(CameraOperation, 'open_device') and not hasattr(obj_cam_operation, 'open_device'): logging.warning("实例缺少open_device方法，手动绑定...") obj_cam_operation.Open_device = types.MethodType(CameraOperation.Open_device, obj_cam_operation) # 确保设备索引有效 if nSelCamIndex >= deviceList.nDeviceNum: QMessageBox.warning(mainWindow, "Error", f"无效的设备索引: {nSelCamIndex}", QMessageBox.Ok) return MV_E_PARAMETER try: ret = obj_cam_operation.Open_device( device_index=nSelCamIndex, access_mode=0 # 使用默认值 ) #ret = obj_cam_operation.Open_device(nSelCamIndex) #ret = obj_cam_operation.Open_device(device_index=nSelCamIndex) except AttributeError as e: logging.error(f"调用open_device失败: {str(e)}") QMessageBox.critical(mainWindow, "严重错误", f"相机操作对象缺少必要方法: {str(e)}", QMessageBox.Ok) return MV_E_CALLORDER if ret != 0: strError = f"打开设备失败 ret: {ToHexStr(ret)}" logging.error(strError) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) isOpen = False else: # 成功打开设备后的操作 logging.info("设备打开成功") set_continue_mode() get_param() isOpen = True enable_controls() # 创建并启动帧监控线程 frame_monitor_thread = FrameMonitorThread(obj_cam_operation) frame_monitor_thread.frame_status.connect(mainWindow.statusBar().showMessage) frame_monitor_thread.start() logging.info("帧监控线程已启动") obj_cam_operation = CameraOperation(cam, deviceList, nSelCamIndex) logging.info(f"CameraOperation 版本: {obj_cam_operation.VERSION}") return ret def start_grabbing(): global obj_cam_operation, isGrabbing # 关键改进：添加相机状态检查 if not obj_cam_operation or not hasattr(obj_cam_operation, 'cam') or not obj_cam_operation.cam: QMessageBox.warning(mainWindow, "Error", "相机对象未正确初始化", QMessageBox.Ok) return ret = obj_cam_operation.Start_grabbing(mainWindow.widgetDisplay.winId()) if ret != 0: strError = "开始取流失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: isGrabbing = True enable_controls() # 等待第一帧到达 QThread.msleep(500) if not obj_cam_operation.is_frame_available(): QMessageBox.warning(mainWindow, "警告", "开始取流后未接收到帧，请检查相机连接！", QMessageBox.Ok) def stop_grabbing(): global obj_cam_operation, isGrabbing # 关键改进：添加相机状态检查 if not obj_cam_operation or not hasattr(obj_cam_operation, 'cam') or not obj_cam_operation.cam: QMessageBox.warning(mainWindow, "Error", "相机对象未正确初始化", QMessageBox.Ok) return # 关键改进：添加连接状态检查 if not hasattr(obj_cam_operation, 'connected') or not obj_cam_operation.connected: QMessageBox.warning(mainWindow, "Error", "相机未连接", QMessageBox.Ok) return ret = obj_cam_operation.Stop_grabbing() if ret != 0: strError = "停止取流失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: isGrabbing = False enable_controls() def close_device(): global isOpen, isGrabbing, obj_cam_operation, frame_monitor_thread if frame_monitor_thread and frame_monitor_thread.isRunning(): frame_monitor_thread.stop() frame_monitor_thread.wait(2000) if isOpen and obj_cam_operation: # 关键改进：确保相机对象存在 if hasattr(obj_cam_operation, 'cam') and obj_cam_operation.cam: obj_cam_operation.Close_device() isOpen = False isGrabbing = False enable_controls() def set_continue_mode(): # 关键改进：添加相机状态检查 if not obj_cam_operation or not hasattr(obj_cam_operation, 'cam') or not obj_cam_operation.cam: return ret = obj_cam_operation.Set_trigger_mode(False) if ret != 0: strError = "设置连续模式失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: mainWindow.radioContinueMode.setChecked(True) mainWindow.radioTriggerMode.setChecked(False) mainWindow.bnSoftwareTrigger.setEnabled(False) def set_software_trigger_mode(): # 关键改进：添加相机状态检查 if not obj_cam_operation or not hasattr(obj_cam_operation, 'cam') or not obj_cam_operation.cam: return ret = obj_cam_operation.Set_trigger_mode(True) if ret != 0: strError = "设置触发模式失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: mainWindow.radioContinueMode.setChecked(False) mainWindow.radioTriggerMode.setChecked(True) mainWindow.bnSoftwareTrigger.setEnabled(isGrabbing) def trigger_once(): # 关键改进：添加相机状态检查 if not obj_cam_operation or not hasattr(obj_cam_operation, 'cam') or not obj_cam_operation.cam: return ret = obj_cam_operation.Trigger_once() if ret != 0: strError = "软触发失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) def save_sample_image(): global isGrabbing, obj_cam_operation, current_sample_path if not isGrabbing: QMessageBox.warning(mainWindow, "错误", "请先开始取流并捕获图像！", QMessageBox.Ok) return # 尝试捕获当前帧 frame = obj_cam_operation.capture_frame() if frame is None: QMessageBox.warning(mainWindow, "无有效图像", "未捕获到有效图像，请检查相机状态！", QMessageBox.Ok) return # 确保图像有效 if frame.size == 0 or frame.shape[0] == 0 or frame.shape[1] == 0: QMessageBox.warning(mainWindow, "无效图像", "捕获的图像无效，请检查相机设置！", QMessageBox.Ok) return settings = QSettings("ClothInspection", "CameraApp") last_dir = settings.value("last_save_dir", os.path.join(os.getcwd(), "captures")) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_filename = f"sample_{timestamp}" file_path, selected_filter = QFileDialog.getSaveFileName( mainWindow, "保存标准样本图像", os.path.join(last_dir, default_filename), "BMP Files (.bmp);;PNG Files (.png);;JPEG Files (.jpg);;所有文件 ()", options=QFileDialog.DontUseNativeDialog ) if not file_path: return # 确保文件扩展名正确 file_extension = os.path.splitext(file_path)[1].lower() if not file_extension: if "BMP" in selected_filter: file_path += ".bmp" elif "PNG" in selected_filter: file_path += ".png" elif "JPEG" in selected_filter or "JPG" in selected_filter: file_path += ".jpg" else: file_path += ".bmp" file_extension = os.path.splitext(file_path)[1].lower() # 创建目录（如果不存在） directory = os.path.dirname(file_path) if directory and not os.path.exists(directory): try: os.makedirs(directory, exist_ok=True) except OSError as e: QMessageBox.critical(mainWindow, "目录创建错误", f"无法创建目录 {directory}: {str(e)}", QMessageBox.Ok) return # 保存图像 try: # 使用OpenCV保存图像 if not cv2.imwrite(file_path, frame): raise Exception("OpenCV保存失败") # 更新状态 current_sample_path = file_path update_sample_display() settings.setValue("last_save_dir", os.path.dirname(file_path)) # 显示成功消息 QMessageBox.information(mainWindow, "成功", f"标准样本已保存至:\n{file_path}", QMessageBox.Ok) # 可选：自动预览样本 preview_sample() except Exception as e: logging.error(f"保存图像失败: {str(e)}") QMessageBox.critical(mainWindow, "保存错误", f"保存图像时发生错误:\n{str(e)}", QMessageBox.Ok) def preview_sample(): global current_sample_path if not current_sample_path or not os.path.exists(current_sample_path): QMessageBox.warning(mainWindow, "错误", "请先设置有效的标准样本图像！", QMessageBox.Ok) return try: # 直接使用OpenCV加载图像 sample_img = cv2.imread(current_sample_path) if sample_img is None: raise Exception("无法加载图像") # 显示图像 cv2.imshow("标准样本预览", sample_img) cv2.waitKey(0) cv2.destroyAllWindows() except Exception as e: QMessageBox.warning(mainWindow, "错误", f"预览样本失败: {str(e)}", QMessageBox.Ok) def start_grabbing(): global obj_cam_operation, isGrabbing ret = obj_cam_operation.Start_grabbing(mainWindow.widgetDisplay.winId()) if ret != 0: strError = "开始取流失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: isGrabbing = True enable_controls() # 等待第一帧到达 QThread.msleep(500) if not obj_cam_operation.is_frame_available(): QMessageBox.warning(mainWindow, "警告", "开始取流后未接收到帧，请检查相机连接！", QMessageBox.Ok) def is_float(str): try: float(str) return True except ValueError: return False def get_param(): try: ret = obj_cam_operation.get_parameters() if ret != MV_OK: strError = "获取参数失败，错误码: " + ToHexStr(ret) QMessageBox.warning(mainWindow, "错误", strError, QMessageBox.Ok) else: mainWindow.edtExposureTime.setText("{0:.2f}".format(obj_cam_operation.exposure_time)) mainWindow.edtGain.setText("{0:.2f}".format(obj_cam_operation.gain)) mainWindow.edtFrameRate.setText("{0:.2f}".format(obj_cam_operation.frame_rate)) except Exception as e: error_msg = f"获取参数时发生错误: {str(e)}" QMessageBox.critical(mainWindow, "严重错误", error_msg, QMessageBox.Ok) def set_param(): frame_rate = mainWindow.edtFrameRate.text() exposure = mainWindow.edtExposureTime.text() gain = mainWindow.edtGain.text() if not (is_float(frame_rate) and is_float(exposure) and is_float(gain)): strError = "设置参数失败: 参数必须是有效的浮点数" QMessageBox.warning(mainWindow, "错误", strError, QMessageBox.Ok) return MV_E_PARAMETER try: ret = obj_cam_operation.set_param( frame_rate=float(frame_rate), exposure_time=float(exposure), gain=float(gain) ) if ret != MV_OK: strError = "设置参数失败，错误码: " + ToHexStr(ret) QMessageBox.warning(mainWindow, "错误", strError, QMessageBox.Ok) except Exception as e: error_msg = f"设置参数时发生错误: {str(e)}" QMessageBox.critical(mainWindow, "严重错误", error_msg, QMessageBox.Ok) def enable_controls(): global isGrabbing, isOpen mainWindow.groupGrab.setEnabled(isOpen) mainWindow.paramgroup.setEnabled(isOpen) mainWindow.bnOpen.setEnabled(not isOpen) mainWindow.bnClose.setEnabled(isOpen) mainWindow.bnStart.setEnabled(isOpen and (not isGrabbing)) mainWindow.bnStop.setEnabled(isOpen and isGrabbing) mainWindow.bnSoftwareTrigger.setEnabled(isGrabbing and mainWindow.radioTriggerMode.isChecked()) mainWindow.bnSaveImage.setEnabled(isOpen and isGrabbing) mainWindow.bnCheckPrint.setEnabled(isOpen and isGrabbing) mainWindow.bnSaveSample.setEnabled(isOpen and isGrabbing) mainWindow.bnPreviewSample.setEnabled(bool(current_sample_path)) def update_sensor_display(data): if not data: return text = (f"张力: {data['tension']:.2f}N | " f"速度: {data['speed']:.2f}m/s | " f"温度: {data['temperature']:.1f}°C | " f"湿度: {data['humidity']:.1f}%") mainWindow.lblSensorData.setText(text) def connect_sensor(): global sensor_monitor_thread, sensor_controller sensor_type = mainWindow.cbSensorType.currentText() if sensor_controller is None: sensor_controller = SensorController() if sensor_type == "串口": config = { 'type': 'serial', 'port': mainWindow.cbComPort.currentText(), 'baudrate': int(mainWindow.cbBaudrate.currentText()), 'timeout': 1.0 } else: config = { 'type': 'ethernet', 'ip': mainWindow.edtIP.text(), 'port': int(mainWindow.edtPort.text()), 'timeout': 1.0 } if sensor_controller.connect(config): mainWindow.bnConnectSensor.setEnabled(False) mainWindow.bnDisconnectSensor.setEnabled(True) sensor_monitor_thread = SensorMonitorThread(sensor_controller) sensor_monitor_thread.data_updated.connect(update_sensor_display) sensor_monitor_thread.start() def disconnect_sensor(): global sensor_monitor_thread if sensor_controller: sensor_controller.disconnect() mainWindow.bnConnectSensor.setEnabled(True) mainWindow.bnDisconnectSensor.setEnabled(False) if sensor_monitor_thread and sensor_monitor_thread.isRunning(): sensor_monitor_thread.stop() sensor_monitor_thread.wait(2000) sensor_monitor_thread = None mainWindow.lblSensorData.setText("传感器数据: 未连接") def update_sensor_ui(index): mainWindow.serialGroup.setVisible(index == 0) mainWindow.ethernetGroup.setVisible(index == 1) class MainWindow(QMainWindow): def init(self): super().init() self.setWindowTitle("布料印花检测系统") self.resize(1200, 800) central_widget = QWidget() self.setCentralWidget(central_widget) main_layout = QVBoxLayout(central_widget) # 设备枚举区域 device_layout = QHBoxLayout() self.ComboDevices = QComboBox() self.bnEnum = QPushButton("枚举设备") self.bnOpen = QPushButton("打开设备") self.bnClose = QPushButton("关闭设备") device_layout.addWidget(self.ComboDevices) device_layout.addWidget(self.bnEnum) device_layout.addWidget(self.bnOpen) device_layout.addWidget(self.bnClose) main_layout.addLayout(device_layout) # 取流控制组 self.groupGrab = QGroupBox("取流控制") grab_layout = QHBoxLayout(self.groupGrab) self.bnStart = QPushButton("开始取流") self.bnStop = QPushButton("停止取流") self.radioContinueMode = QRadioButton("连续模式") self.radioTriggerMode = QRadioButton("触发模式") self.bnSoftwareTrigger = QPushButton("软触发") grab_layout.addWidget(self.bnStart) grab_layout.addWidget(self.bnStop) grab_layout.addWidget(self.radioContinueMode) grab_layout.addWidget(self.radioTriggerMode) grab_layout.addWidget(self.bnSoftwareTrigger) main_layout.addWidget(self.groupGrab) # 参数设置组 self.paramgroup = QGroupBox("相机参数") param_layout = QGridLayout(self.paramgroup) self.edtExposureTime = QLineEdit() self.edtGain = QLineEdit() self.edtFrameRate = QLineEdit() self.bnGetParam = QPushButton("获取参数") self.bnSetParam = QPushButton("设置参数") self.bnSaveImage = QPushButton("保存图像") param_layout.addWidget(QLabel("曝光时间:"), 0, 0) param_layout.addWidget(self.edtExposureTime, 0, 1) param_layout.addWidget(self.bnGetParam, 0, 2) param_layout.addWidget(QLabel("增益:"), 1, 0) param_layout.addWidget(self.edtGain, 1, 1) param_layout.addWidget(self.bnSetParam, 1, 2) param_layout.addWidget(QLabel("帧率:"), 2, 0) param_layout.addWidget(self.edtFrameRate, 2, 1) param_layout.addWidget(self.bnSaveImage, 2, 2) main_layout.addWidget(self.paramgroup) # 图像显示区域 self.widgetDisplay = QLabel() self.widgetDisplay.setMinimumSize(640, 480) self.widgetDisplay.setStyleSheet("background-color: black;") self.widgetDisplay.setAlignment(Qt.AlignCenter) self.widgetDisplay.setText("相机预览区域") main_layout.addWidget(self.widgetDisplay, 1) # 状态栏 #self.statusBar = QStatusBar() #self.setStatusBar(self.statusBar) # 创建自定义UI组件 self.setup_custom_ui() def setup_custom_ui(self): # 工具栏 toolbar = self.addToolBar("检测工具") self.bnCheckPrint = QPushButton("检测印花质量") self.bnSaveSample = QPushButton("保存标准样本") self.bnPreviewSample = QPushButton("预览样本") self.cbHistory = QComboBox() self.cbHistory.setMinimumWidth(300) toolbar.addWidget(self.bnCheckPrint) toolbar.addWidget(self.bnSaveSample) toolbar.addWidget(self.bnPreviewSample) toolbar.addWidget(QLabel("历史记录:")) toolbar.addWidget(self.cbHistory) # 状态栏样本路径 self.lblSamplePath = QLabel("当前样本: 未设置样本") self.statusBar().addPermanentWidget(self.lblSamplePath) # 右侧面板 right_panel = QWidget() right_layout = QVBoxLayout(right_panel) right_layout.setContentsMargins(10, 10, 10, 10) # 差异度调整组 diff_group = QGroupBox("差异度调整") diff_layout = QVBoxLayout(diff_group) self.lblDiffThreshold = QLabel("差异度阈值 (0-100%):") self.sliderDiffThreshold = QSlider(Qt.Horizontal) self.sliderDiffThreshold.setRange(0, 100) self.sliderDiffThreshold.setValue(5) self.lblDiffValue = QLabel("5%") self.lblCurrentDiff = QLabel("当前差异度: -") self.lblCurrentDiff.setStyleSheet("font-size: 14px; font-weight: bold;") self.lblDiffStatus = QLabel("状态: 未检测") self.lblDiffStatus.setStyleSheet("font-size: 12px;") diff_layout.addWidget(self.lblDiffThreshold) diff_layout.addWidget(self.sliderDiffThreshold) diff_layout.addWidget(self.lblDiffValue) diff_layout.addWidget(self.lblCurrentDiff) diff_layout.addWidget(self.lblDiffStatus) right_layout.addWidget(diff_group) # 传感器控制面板 sensor_panel = QGroupBox("传感器控制") sensor_layout = QVBoxLayout(sensor_panel) sensor_type_layout = QHBoxLayout() self.lblSensorType = QLabel("传感器类型:") self.cbSensorType = QComboBox() self.cbSensorType.addItems(["串口", "以太网"]) sensor_type_layout.addWidget(self.lblSensorType) sensor_type_layout.addWidget(self.cbSensorType) sensor_layout.addLayout(sensor_type_layout) # 串口参数 self.serialGroup = QGroupBox("串口参数") serial_layout = QVBoxLayout(self.serialGroup) self.lblComPort = QLabel("端口:") self.cbComPort = QComboBox() if platform.system() == 'Windows': ports = [f"COM{i}" for i in range(1, 21)] else: ports = [f"/dev/ttyS{i}" for i in range(0, 4)] + [f"/dev/ttyUSB{i}" for i in range(0, 4)] self.cbComPort.addItems(ports) self.lblBaudrate = QLabel("波特率:") self.cbBaudrate = QComboBox() self.cbBaudrate.addItems(["96000", "19200", "38400", "57600", "115200"]) self.cbBaudrate.setCurrentText("115200") serial_layout.addWidget(self.lblComPort) serial_layout.addWidget(self.cbComPort) serial_layout.addWidget(self.lblBaudrate) serial_layout.addWidget(self.cbBaudrate) sensor_layout.addWidget(self.serialGroup) # 以太网参数 self.ethernetGroup = QGroupBox("以太网参数") ethernet_layout = QVBoxLayout(self.ethernetGroup) self.lblIP = QLabel("IP地址:") self.edtIP = QLineEdit("192.168.1.100") self.lblPort = QLabel("端口:") self.edtPort = QLineEdit("502") ethernet_layout.addWidget(self.lblIP) ethernet_layout.addWidget(self.edtIP) ethernet_layout.addWidget(self.lblPort) ethernet_layout.addWidget(self.edtPort) sensor_layout.addWidget(self.ethernetGroup) # 连接按钮 self.bnConnectSensor = QPushButton("连接传感器") self.bnDisconnectSensor = QPushButton("断开传感器") self.bnDisconnectSensor.setEnabled(False) sensor_layout.addWidget(self.bnConnectSensor) sensor_layout.addWidget(self.bnDisconnectSensor) # 延迟设置 delay_layout = QHBoxLayout() self.lblDelay = QLabel("触发延迟(秒):") self.spinDelay = QSpinBox() self.spinDelay.setRange(0, 60) self.spinDelay.setValue(0) self.spinDelay.setToolTip("传感器检测到布料后延迟拍摄的时间") delay_layout.addWidget(self.lblDelay) delay_layout.addWidget(self.spinDelay) sensor_layout.addLayout(delay_layout) # 传感器数据 self.lblSensorData = QLabel("传感器数据: 未连接") self.lblSensorData.setStyleSheet("font-size: 10pt;") sensor_layout.addWidget(self.lblSensorData) right_layout.addWidget(sensor_panel) right_layout.addStretch(1) # 停靠窗口 dock = QDockWidget("检测控制面板", self) dock.setWidget(right_panel) dock.setFeatures(QDockWidget.DockWidgetMovable | QDockWidget.DockWidgetFloatable) self.addDockWidget(Qt.RightDockWidgetArea, dock) def closeEvent(self, event): logging.info("主窗口关闭，执行清理...") close_device() disconnect_sensor() event.accept() if name == "main": path_fixer = PathFixer() app = QApplication(sys.argv) mainWindow = MainWindow() # 信号连接 mainWindow.cbSensorType.currentIndexChanged.connect(update_sensor_ui) update_sensor_ui(0) mainWindow.bnConnectSensor.clicked.connect(connect_sensor) mainWindow.bnDisconnectSensor.clicked.connect(disconnect_sensor) mainWindow.sliderDiffThreshold.valueChanged.connect(update_diff_threshold) mainWindow.bnCheckPrint.clicked.connect(sensor_controlled_check) mainWindow.bnSaveSample.clicked.connect(save_sample_image) mainWindow.bnPreviewSample.clicked.connect(preview_sample) mainWindow.bnEnum.clicked.connect(enum_devices) mainWindow.bnOpen.clicked.connect(open_device) mainWindow.bnClose.clicked.connect(close_device) mainWindow.bnStart.clicked.connect(start_grabbing) mainWindow.bnStop.clicked.connect(stop_grabbing) mainWindow.bnSoftwareTrigger.clicked.connect(trigger_once) mainWindow.radioTriggerMode.clicked.connect(set_software_trigger_mode) mainWindow.radioContinueMode.clicked.connect(set_continue_mode) mainWindow.bnGetParam.clicked.connect(get_param) mainWindow.bnSetParam.clicked.connect(set_param) mainWindow.bnSaveImage.clicked.connect(save_sample_image) main() mainWindow.show() app.exec_() close_device() disconnect_sensor() sys.exit() 这是目前主程序的的代码，出现了下面的错误 Traceback (most recent call last): File "c:\Users\zhangyuxiang\Desktop\cloth\main\clothing_inspection.py", line 56, in <module> from mv_import.Cam_Operation_class import CameraOperation File "c:\Users\zhangyuxiang\Desktop\cloth\main\mv_import\Cam_Operation_class.py", line 11, in <module> from .MvCameraControl_class import * File "c:\Users\zhangyuxiang\Desktop\cloth\main\mv_import\MvCameraControl_class.py", line 18, in <module> from .Camera_Params_header import * File "c:\Users\zhangyuxiang\Desktop\cloth\main\mv_import\Camera_Params_header.py", line 5, in <module> from .CameraParams_const import * ModuleNotFoundError: No module named 'mv_import.CameraParams_const'

3. **包结构问题**：mv_import目录缺少__init__.py文件 4. **命名冲突**：文件名大小写不匹配（Windows不区分大小写，但Python导入区分） 5. **导入方式错误**：相对导入在非包环境中使用 ### 解决方案 #### 1...

ERROR - Traceback (most recent call last): File "/tmp/jqcore/jqboson/jqboson/core/entry.py", line 379, in _run engine.start() File "/tmp/jqcore/jqboson/jqboson/core/engine.py", line 246, in start self._dispatcher.start() File "/tmp/jqcore/jqboson/jqboson/core/dispatcher.py", line 280, in start self._run_loop() File "/tmp/jqcore/jqboson/jqboson/core/dispatcher.py", line 240, in _run_loop self._loop.run() File "/tmp/jqcore/jqboson/jqboson/core/loop/loop.py", line 120, in run self._handle_queue() File "/tmp/jqcore/jqboson/jqboson/core/loop/loop.py", line 166, in _handle_queue message.callback(message.callback_data) File "/tmp/jqcore/jqboson/jqboson/core/dispatcher.py", line 110, in callback self._event_bus.emit(evt) File "/tmp/jqcore/jqboson/jqboson/core/bus.py", line 47, in emit ret.append(call(event)) File "/tmp/jqcore/jqboson/jqboson/core/strategy.py", line 433, in _wrapper return cb(self._context.strategy_environment.strategy_context, cb_kwargs) File "/tmp/strategy/user_code.py", line 50, in before_market_open contracts = get_future_contracts(g.future_type) File "/tmp/strategy/user_code.py", line 250, in get_future_contracts return get_dominant_future(future_type, context.current_dt.date()) NameError: name 'context' is not defined报错

另外，在initialize函数中，我们设置run_daily时，before_market_open函数被调用时会传入context，因此我们可以通过context获取当前日期。同时，在trade函数中，我们并没有直接调用get_future_...

根据上述思路写一个聚宽策略

set_params(context) # 参数初始化 set_variables(context) # 变量初始化 set_backtest(context) # 回测设置 def set_params(context): g.ma_period = 34 # 均线周期 g.atr_period = 14 # ATR周期 g.kdj_...

用聚宽写一个etf轮动的策略

set_params(context) # 设置参数 set_backtest() # 设置回测条件 def set_params(context): # ETF池（需替换为实际ETF代码） context.etf_list = [ '510300.XSHG', # 沪深300ETF '510500.XSHG', # 中证500ETF...

在之前的聚宽策略代码的基础上修改即可

from jqdata import * import talib import numpy as np def initialize(context): # ====== 策略全局变量 ====== g.future_code = 'IM8888.CCFX' # IM主力连续合约 g.holding_position = False # 持仓状态 g....

我现在想将仓位控制系统：根据ATR计算每手风险单笔风险≤账户2%，总仓位≤账户8%，单合约≤50手动态计算合理开仓手数，实现了双层止损机制（初始固定止损+移动吊灯止损）删除，并且每开仓只开一手

# 初始化参数函数，设置一些全局变量 def init_params(context): # 定义不同期货品种对应的现货指数和主力合约 ref_dict = { 'IM': ['000852.XSHG', 'IM8888.CCFX'], # 期货连续合约 } g.future_type = 'IM' ...

相关推荐

LK-GD500 initialize_labview基恩士参数初始化_基恩士_

identity_cache_fetchers:用于 IdentityCache 的 ActiveRecord#fetch_or_initialize 和 ActiveRecord#fetch_or_create

LK-GD500-initialize.zip_LABVIEW 基恩士_LK-_LK-GD500设置_labview基恩士_恩基

量化风险管理：JQData在风险预测与暴露分析中的应用

【量化交易与算法交易策略】算法交易策略开发：交易规则的编程实现

【AI训练平台高效能】：Horovod多机多卡训练策略

【Tornado_2.2插件开发】：自定义框架功能的权威指南

Fama French五因子模型的Python自动化回测指南：高效工具与框架介绍

深度学习基础：介绍MXNet框架及其使用

【MXNet开发者指南】：快速掌握Horovod集成

证券分析革命：Python基本面与技术面分析实战

根据上述思路写一个聚宽策略

用聚宽写一个etf轮动的策略

在之前的聚宽策略代码的基础上修改即可

我现在想将仓位控制系统： 根据ATR计算每手风险 单笔风险≤账户2%，总仓位≤账户8%，单合约≤50手 动态计算合理开仓手数，实现了双层止损机制（初始固定止损+移动吊灯止损）删除，并且每开仓只开一手

大家在看

HCIE DC数据中心认证培训教材

04_Human activity recognition based on transformed accelerometer data from a mobile phone

基于ADS的微带滤波器设计

虚拟光驱DAEMON（支持2000/XP/2003）

佳博打印机编程手册esc tspl cpcl

最新推荐

Comsol声子晶体能带计算：六角与三角晶格原胞选取及布里渊区高对称点选择 - 声子晶体 v1.0

springboot213大学生心理健康管理系统的设计与实现.zip

三轴自动锁螺丝机PLC配方编程：吸钉式锁螺丝智能调整与注释详解 变址寄存器 高效版

基于QT与STM32的Modbus-TCP四遥功能实现及源码解析

ERP系统客户与供应商信息视图创建：Oracle数据库中客户和供应商数据整合查询设计

Web前端开发：CSS与HTML设计模式深入解析

Zotero 7数据同步：Attanger插件安装&设置，打造文献管理利器

卷积神经网络的基础理论200字

轻便实用的Java库类查询工具介绍

【Zotero 7终极指南】：新手必备！Attanger插件全攻略与数据同步神技

我现在想将仓位控制系统：根据ATR计算每手风险单笔风险≤账户2%，总仓位≤账户8%，单合约≤50手动态计算合理开仓手数，实现了双层止损机制（初始固定止损+移动吊灯止损）删除，并且每开仓只开一手

三轴自动锁螺丝机PLC配方编程：吸钉式锁螺丝智能调整与注释详解变址寄存器高效版