Forem: drake

加密货币协整检验算法

drake — Wed, 14 Jan 2026 16:50:46 +0000


    @staticmethod
    def _calculate_cointegration_params(base_prices: pd.Series, alt_prices: pd.Series,
                                        coin: str = None, base_symbol: str = None) -> Optional[dict]:
        """
        使用OLS回归计算协整参数（验证性函数）

        通过OLS回归计算截距项α和斜率β，并进行ADF检验验证价差平稳性。
        这是协整检验的标准方法（Engle-Granger两步法）。

        Args:
            base_prices: 基准币种价格序列（pandas Series）
            alt_prices: 山寨币价格序列（pandas Series）
            coin: 币种名称（可选，用于日志）

        Returns:
            dict: {
                'alpha': 截距项（价格溢价/折价）,
                'beta': OLS回归系数（对冲比例）,
                'spread': OLS价差序列（残差，保留原始索引）,
                'adf_pvalue': ADF检验p值（平稳性）
            }
            None: 如果计算失败

        Note:
            - α显著非0表示存在固定溢价/折价
            - β是最优对冲比例
            - ADF p-value < 0.05 表示价差平稳，适合配对交易
            - 此函数用于验证性分析，使用全样本数据是标准做法（Engle-Granger两步法）
            - 返回的价差序列保留原始时间索引，便于后续分析
            - 注意：此函数存在 look-ahead bias（使用全样本），仅用于事后验证，不适用于实时交易
              实时交易应使用 price_diff_spread_ols_window 方法
        """
        try:
            # 1. 数据验证
            if len(base_prices) != len(alt_prices):
                coin_info = f" | 币种: {coin}" if coin else ""
                base_symbol_info = f" | 基准币种: {base_symbol}" if base_symbol else ""
                logger.warning(f"协整参数计算失败：基准币种和ALT数据长度不一致 | "
                              f"基准币种: {len(base_prices)}, ALT: {len(alt_prices)}"
                              f"{coin_info}{base_symbol_info}")
                return None

            if len(base_prices) < 10:  # 最小数据点要求
                coin_info = f" | 币种: {coin}" if coin else ""
                logger.debug(f"协整参数计算失败：数据点不足（需要至少10个点，实际{len(base_prices)}个）{coin_info}")
                return None

            # 2. 计算对数价格（保留索引信息）
            log_base_series = np.log(base_prices)
            log_alt_series = np.log(alt_prices)

            # 3. statsmodels OLS回归（带常数项）：log_alt = α + β * log_base + ε
            # 注意：此函数用于验证性分析，使用全样本是标准做法（Engle-Granger两步法）
            # 虽然存在 look-ahead bias，但这是协整检验的标准方法
            X = sm.add_constant(log_base_series)
            model = sm.OLS(log_alt_series, X).fit()

            alpha = model.params.iloc[0]      # 常数项
            beta = model.params.iloc[1]       # 斜率
            alpha_pvalue = model.pvalues.iloc[0]  # α的p值
            beta_pvalue = model.pvalues.iloc[1]   # β的p值
            rsquared = model.rsquared    # 拟合优度

            # 4. 根据α显著性和绝对值大小选择价差计算方法（智能模型选择）
            if alpha_pvalue < 0.05 and abs(alpha) > 5:
                # α显著且绝对值很大 → 跨资产类配对（如NEAR/BTC）
                # 使用无α模型更稳健（避免α时变性问题）
                spread_ols = log_alt_series - beta * log_base_series
                model_type = "no_intercept_forced"
                use_alpha = False
                model_reason = f"|α|={abs(alpha):.1f}>5, 跨资产类配对"

            elif alpha_pvalue < 0.05 and abs(alpha) < 2:
                # α显著且绝对值较小 → 同类资产配对（如UNI/SUSHI）
                # α代表真实的溢价关系，应当包含
                spread_ols = log_alt_series - (alpha + beta * log_base_series)
                model_type = "standard_EG"  # 标准Engle-Granger
                use_alpha = True
                model_reason = f"|α|={abs(alpha):.1f}<2, 同类资产配对"

            else:
                # α不显著或中等范围（2<=|α|<=5）→ 使用无α模型
                spread_ols = log_alt_series - beta * log_base_series
                model_type = "no_intercept"
                use_alpha = False
                if alpha_pvalue >= 0.05:
                    model_reason = "α不显著"
                else:
                    model_reason = f"|α|={abs(alpha):.1f}∈[2,5], 中等范围"

            # 5. ADF检验价差平稳性（使用数值数组）
            adf_result = adfuller(spread_ols.values, autolag='AIC')
            adf_pvalue = adf_result[1]

            # 6. 日志输出（便于调试）
            if coin:
                logger.debug(
                    f"协整参数 | 币种: {coin} | α={alpha:.4f} (p={alpha_pvalue:.4f}) | "
                    f"β={beta:.4f} (p={beta_pvalue:.4f}) | R²={rsquared:.4f} | "
                    f"模型: {model_type} | 原因: {model_reason} | ADF p={adf_pvalue:.4f}"
                )

            return {
                'alpha': alpha,
                'beta': beta,
                'spread': spread_ols,  # 保留原始索引的 pandas Series
                'adf_pvalue': adf_pvalue,
                # 新增统计信息
                'alpha_pvalue': alpha_pvalue,
                'beta_pvalue': beta_pvalue,
                'rsquared': rsquared,
                'model_type': model_type,
                'use_alpha': use_alpha,  # 标记是否使用了α
                'model_reason': model_reason  # 模型选择原因
            }
        except Exception as e:
            coin_info = f" | 币种: {coin}" if coin else ""
            logger.debug(f"OLS协整参数计算失败：{type(e).__name__}: {str(e)}{coin_info}", exc_info=True)
            return None

    @staticmethod
    def price_diff_spread_ols_window(base_prices: pd.Series, alt_prices: pd.Series, beta_window: int = 100, zscore_window: int = 30) -> pd.Series:
        """
        计算价格差价（双窗口策略：OLS回归使用长窗口（稳定），统计量使用短窗口（敏感）。）
        """
        # 4. 数据切片：取足够计算OLS回归和统计量的数据
        data_window = max(beta_window, zscore_window)
        recent_base_full = base_prices.iloc[-data_window:]
        recent_alt_full = alt_prices.iloc[-data_window:]

        # 5. OLS回归计算协整参数
        # 使用前 beta_window-1 个点计算OLS参数（避免 look-ahead bias）
        # 公式：log_alt = α + β × log_base + ε
        # 用途：构建价差序列 spread = log(ALT) - (α + β × log(BASE))
        ols_base = recent_base_full.iloc[:-1]
        ols_alt = recent_alt_full.iloc[:-1]

        # 计算对数价格
        log_base_ols = np.log(ols_base)
        log_alt_ols = np.log(ols_alt)

        # statsmodels OLS回归
        X = sm.add_constant(log_base_ols)
        model = sm.OLS(log_alt_ols, X).fit()

        alpha = model.params.iloc[0]          # 常数项
        beta_ols = model.params.iloc[1]       # 斜率
        alpha_pvalue = model.pvalues.iloc[0]  # α的p值
        beta_pvalue = model.pvalues.iloc[1]   # β的p值
        rsquared = model.rsquared         # 拟合优度

        # 根据α显著性和绝对值大小选择价差计算方法（智能模型选择）
        log_base_full = np.log(recent_base_full)  # 全部100期
        log_alt_full = np.log(recent_alt_full)

        if alpha_pvalue < 0.05 and abs(alpha) > 5:
            # α显著且绝对值很大 → 跨资产类配对（如NEAR/BTC）
            # 使用无α模型更稳健（避免α时变性问题）
            spread_full = log_alt_full - beta_ols * log_base_full
            model_type = "no_intercept_forced"
            use_alpha = False
            model_reason = f"|α|={abs(alpha):.1f}>5, 跨资产类配对"

        elif alpha_pvalue < 0.05 and abs(alpha) < 2:
            # α显著且绝对值较小 → 同类资产配对（如UNI/SUSHI）
            # α代表真实的溢价关系，应当包含
            spread_full = log_alt_full - (alpha + beta_ols * log_base_full)
            model_type = "standard_EG"
            use_alpha = True
            model_reason = f"|α|={abs(alpha):.1f}<2, 同类资产配对"

        else:
            # α不显著或中等范围（2<=|α|<=5）→ 使用无α模型
            spread_full = log_alt_full - beta_ols * log_base_full
            model_type = "no_intercept"
            use_alpha = False
            if alpha_pvalue >= 0.05:
                model_reason = "α不显著"
            else:
                model_reason = f"|α|={abs(alpha):.1f}∈[2,5], 中等范围"

        # ADF检验价差平稳性
        adf_result = adfuller(spread_full.values, autolag='AIC')
        adf_pvalue = adf_result[1]

        # 6. 价差构建（用于Z-score计算：使用短窗口保持敏感度）
        # 取最近 zscore_window 期数据，使用长窗口计算的OLS参数构建对数价差
        recent_base = recent_base_full.iloc[-zscore_window:]
        recent_alt = recent_alt_full.iloc[-zscore_window:]
        log_base = np.log(recent_base)
        log_alt = np.log(recent_alt)

        # 使用相同的模型选择
        if use_alpha:
            spread = log_alt - (alpha + beta_ols * log_base)
        else:
            spread = log_alt - beta_ols * log_base

        return {
            'alpha': alpha,           # 截距项（价格溢价/折价）
            'beta': beta_ols,         # OLS回归系数
            'spread': spread,         # 用于Z-score计算的价差序列
            'adf_pvalue': adf_pvalue, # ADF检验价差平稳性
            # 新增统计信息
            'alpha_pvalue': alpha_pvalue,
            'beta_pvalue': beta_pvalue,
            'rsquared': rsquared,
            'model_type': model_type,
            'use_alpha': use_alpha,
            'model_reason': model_reason  # 模型选择原因
        }

    def cointegration_analysis(
        self, cointegration_result: dict, 
        method_type: str, 
        coin: str = None, 
        stats_period_key: tuple = None
        ) -> bool:
        """
        协整分析
        """
        # 协整检验状态
        cointegration_status = False
        # 币种信息
        coin_info = f" | 币种: {coin} | 方法: {method_type}" if coin else ""
        # check_result = ''
        if cointegration_result is None or cointegration_result['adf_pvalue'] >= 0.05:
            if cointegration_result:
                adf_pvalue_str = f"{cointegration_result['adf_pvalue']:.4f}"
                alpha_str = f"{cointegration_result['alpha']:.4f}"
                beta_str = f"{cointegration_result['beta']:.4f}"
                check_result = (f"❌ 协整检验未通过（基于{stats_period_key}周期数据） | "
                                f"α={alpha_str}, β={beta_str} | "
                                f"ADF p-value: {adf_pvalue_str} >= 0.05 | "
                                f"原因: 价差非平稳，不适合配对交易"
                                f"{coin_info}")
            else:
                check_result = (
                    f"❌ 协整检验未通过（基于{stats_period_key}周期数据） | "
                    f"ADF p-value: N/A | "
                    f"原因: OLS参数计算失败"
                    f"{coin_info}"
                )
            # is_anomaly = False  # ⚠️ 协整失败，拒绝信号
        else:
            # 协整检验通过，输出详细信息
            check_result = (f"✅ 协整检验通过（基于{stats_period_key}周期数据） | "
                            f"α={cointegration_result['alpha']:.4f}, β={cointegration_result['beta']:.4f} | "
                            f"ADF p-value={cointegration_result['adf_pvalue']:.4f} < 0.05"
                            f"{coin_info}")
            cointegration_status = True
        self.alert_content += f"\n{check_result}\n"
        logger.info(check_result)

        return cointegration_status

tailscaled 保活 shell 脚本

drake — Tue, 09 Dec 2025 11:36:44 +0000

1.创建文件

touch tailscaled_monitor.sh

2. 编辑文件

#!/bin/bash

# tailscaled 进程监控脚本
# 功能：检测 tailscaled 进程是否存活，如果死掉则自动重启

LOG_FILE="/var/log/tailscaled_monitor.log"
CHECK_INTERVAL=10  # 检测间隔（秒）

# 日志函数
log_message() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_FILE"
}

# 启动 tailscaled 函数
start_tailscaled() {
    log_message "检测到 tailscaled 进程已停止，正在重启..."

    tailscaled \
        --state=/var/lib/tailscale/tailscaled.state \
        --socket=/var/run/tailscale/tailscaled.sock \
        --tun=userspace-networking &

    sleep 2

    if pgrep -f "tailscaled.*--state=/var/lib/tailscale/tailscaled.state" > /dev/null; then
        log_message "tailscaled 进程重启成功 (PID: $(pgrep -f 'tailscaled.*--state=/var/lib/tailscale/tailscaled.state'))"
    else
        log_message "错误：tailscaled 进程重启失败"
    fi
}

# 检查 tailscaled 进程
check_tailscaled() {
    # 使用 pgrep 查找包含特定参数的 tailscaled 进程
    if pgrep -f "tailscaled.*--state=/var/lib/tailscale/tailscaled.state" > /dev/null; then
        return 0  # 进程存在
    else
        return 1  # 进程不存在
    fi
}

log_message "tailscaled 监控脚本启动"

# 主循环
while true; do
    if ! check_tailscaled; then
        start_tailscaled
    fi

    sleep "$CHECK_INTERVAL"
done

3. 赋予文件执行权限

chmod +x tailscaled_monitor.sh

4. 后台运行脚本

nohup ./tailscaled_monitor.sh > /dev/null 2>&1 &

5. 校验脚本是否运行成功（查看日志）

tail -f /var/log/tailscaled_monitor.log

akash DSL 参考

drake — Tue, 09 Dec 2025 04:54:59 +0000

---
version: "2.0"
services:
  service-1:
    image: ghcr.io/akash-network/ubuntu-2404-ssh:2
    command: ["/bin/sh", "-c"]
    args:
      - |
        mkdir -p /var/lib/tailscale /var/run/tailscale

        tailscaled \
          --tun=userspace-networking \
          --state=/var/lib/tailscale/tailscaled.state \
          --socket=/var/run/tailscale/tailscaled.sock &

        until tailscale --socket /var/run/tailscale/tailscaled.sock status 2>/dev/null | grep -q "100."; do
          sleep 2
        done

        if ! tailscale --socket /var/run/tailscale/tailscaled.sock status | grep -q "logged in"; then
          tailscale --socket /var/run/tailscale/tailscaled.sock up \
            --auth-key=tskey-auth-k2RyTfkfqB11CNTRL-ZxbbXL4c4wWV9q5TDjTcvWbg1g1hhHCRA \
            --accept-dns=false \
            --accept-routes=false
        fi

        echo "Tailscale is running at $(tailscale --socket /var/run/tailscale/tailscaled.sock ip -4)"

        exec /entrypoint.sh   # 启动原来的 SSH 服务

    expose:
      - port: 22
        as: 22
        to:
          - global: false      # 现在可以彻底关掉公网 SSH 了！
      - port: 80
        as: 80
        to:
          - global: false      # 公网 80 也关掉，只走 Tailscale 访问

    env:
      - SSH_PUBKEY=ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIJxcN3WQAU6KUS/13R9Wt5/dlA3YKHoAX48pHUjt8K3w shixiaolong09@gmail.com

profiles:
  compute:
    service-1:
      resources:
        cpu: { units: 0.5 }
        memory: 1Gi
        storage: [{ size: 2Gi }]
  placement:
    dcloud:
      pricing:
        service-1:
          denom: uakt
          amount: 5000
deployment:
  service-1:
    dcloud:
      profile: service-1
      count: 1

容器内安装tailscale

drake — Tue, 09 Dec 2025 01:24:58 +0000

1.安装

curl -fsSL https://tailscale.com/install.sh | sh && tailscale up --auth-key=tskey-auth-kxxxx

2.启动

tailscaled \
  --state=/var/lib/tailscale/tailscaled.state \
  --socket=/var/run/tailscale/tailscaled.sock \
  --tun=userspace-networking &

3.登录

tailscale --socket /var/run/tailscale/tailscaled.sock up \
  --auth-key=tskey-auth-k2RxxxxxxA \
  --accept-dns=false \
  --accept-routes=false

4.校验是否成功

tailscale --socket /var/run/tailscale/tailscaled.sock status
tailscale --socket /var/run/tailscale/tailscaled.sock ip -4

5. 设置为网络出口

tailscale set --advertise-exit-node

6. 重启、启动（掉线了或者需要重启）

tailscale up

使用 faster-whisper批量转录的python脚本

drake — Thu, 04 Dec 2025 02:02:00 +0000

"""
批量转录当前目录下的 .mp3 文件，使用 faster-whisper
转录完成后立即删除原始 .mp3 文件，仅保留 .txt
"""
import os
import sys
import logging
import argparse
from pathlib import Path
from faster_whisper import WhisperModel

# ================== 日志配置 ==================
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('Extract Text')

# ================== 配置区 ==================
MODEL_SIZE = "base"        # 可选: tiny, base, small, medium, large
DEVICE = "cpu"              # cpu 或 cuda
COMPUTE_TYPE = "int8"       # int8, float16, float32 (CPU 推荐 int8)
VAD_FILTER = True           # 启用语音活动检测，去除静音
OUTPUT_FORMAT = "txt"       # 输出格式
VERBOSE = True              # 是否显示详细日志

# ===========================================

def transcribe_audio(audio_path: Path, model: WhisperModel) -> str:
    """转录单个音频文件，返回文本内容"""
    print(f"转录: {audio_path.name} → {audio_path.stem}.txt")

    segments, info = model.transcribe(
        str(audio_path),
        language=None,
        beam_size=5,
        vad_filter=VAD_FILTER,
        vad_parameters=dict(min_silence_duration_ms=500),
        word_timestamps=False,
    )

    text_lines = []
    for segment in segments:
        line = segment.text.strip()
        if line:
            text_lines.append(line)
        if VERBOSE:
            logger.info(f"[{segment.start:06.2f}s --> {segment.end:06.2f}s] {line}")


    return "\n".join(text_lines)


def safe_delete(file_path: Path):
    """安全删除文件，捕获异常并记录"""
    try:
        file_path.unlink()
        logger.info(f"已删除: {file_path.name}")
    except Exception as e:
        logger.error(f"删除失败 {file_path.name}: {e}")


def main(keep_audio: bool = False):
    print("=== faster-whisper 批量转录 + 自动清理音频 ===")

    current_dir = Path(".")
    mp3_files = sorted(current_dir.glob("*.mp3"))

    if not mp3_files:
        print("未找到 .mp3 文件，退出。")
        return

    # 加载模型（只加载一次）
    print(f"正在加载模型 {MODEL_SIZE} ({DEVICE}, {COMPUTE_TYPE})...")
    model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE)

    processed = 0
    for mp3_path in mp3_files:
        txt_path = mp3_path.with_suffix(".txt")

        # 如果 .txt 已存在，跳过（但仍可选择删除 .mp3）
        if txt_path.exists():
            print(f"跳过: {txt_path.name} 已存在")
            if not keep_audio and mp3_path.exists():
                safe_delete(mp3_path)
            continue

        try:
            # 转录
            text = transcribe_audio(mp3_path, model)
            if not text.strip():
                print(f"警告: {mp3_path.name} 转录结果为空，跳过删除")
                safe_delete(mp3_path)
                continue

            # 写入文本
            txt_path.write_text(text, encoding="utf-8")
            processed += 1
            print(f"已保存: {txt_path.name}")

            # 删除原始音频（仅当不保留时）
            if not keep_audio:
                safe_delete(mp3_path)

        except Exception as e:
            print(f"错误转录 {mp3_path.name}: {e}", file=sys.stderr)
            logger.error(f"处理失败: {mp3_path.name} - {e}")

    mode = "保留音频" if keep_audio else "已删除音频"
    print(f"全部完成！共处理 {processed} 个文件，{mode}。")


if __name__ == "__main__":
    main()

NVIDIA GeForce GTX 750 Ti 跑 faster-whisper

drake — Fri, 14 Nov 2025 02:25:39 +0000

"""
批量转录当前目录下的 .mp3 文件，使用 faster-whisper
转录完成后立即删除原始 .mp3 文件，仅保留 .txt
"""
import os
import sys
import logging
import argparse
from pathlib import Path
from faster_whisper import WhisperModel

# ================== 日志配置 ==================
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('Extract Text')

# ================== 配置区 ==================
MODEL_SIZE = "base"        # 可选: tiny, base, small, medium, large
DEVICE = "cuda"              # cpu 或 cuda
COMPUTE_TYPE = "float32"       # int8, float16, float32 (CPU 推荐 int8)
VAD_FILTER = True           # 启用语音活动检测，去除静音
OUTPUT_FORMAT = "txt"       # 输出格式
VERBOSE = True              # 是否显示详细日志

# ===========================================

def transcribe_audio(audio_path: Path, model: WhisperModel) -> str:
    """转录单个音频文件，返回文本内容"""
    print(f"转录: {audio_path.name} → {audio_path.stem}.txt")

    segments, info = model.transcribe(
        str(audio_path),
        language=None,
        beam_size=5,
        vad_filter=VAD_FILTER,
        vad_parameters=dict(min_silence_duration_ms=500),
        word_timestamps=False,
    )

    text_lines = []
    for segment in segments:
        line = segment.text.strip()
        if line:
            text_lines.append(line)
        if VERBOSE:
            logger.info(f"[{segment.start:06.2f}s --> {segment.end:06.2f}s] {line}")


    return "\n".join(text_lines)


def safe_delete(file_path: Path):
    """安全删除文件，捕获异常并记录"""
    try:
        file_path.unlink()
        logger.info(f"已删除: {file_path.name}")
    except Exception as e:
        logger.error(f"删除失败 {file_path.name}: {e}")


def main(keep_audio: bool = False):
    print("=== faster-whisper 批量转录 + 自动清理音频 ===")

    current_dir = Path(".")
    mp3_files = sorted(current_dir.glob("*.mp3"))

    if not mp3_files:
        print("未找到 .mp3 文件，退出。")
        return

    # 加载模型（只加载一次）
    print(f"正在加载模型 {MODEL_SIZE} ({DEVICE}, {COMPUTE_TYPE})...")
    model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE)

    processed = 0
    for mp3_path in mp3_files:
        txt_path = mp3_path.with_suffix(".txt")

        # 如果 .txt 已存在，跳过（但仍可选择删除 .mp3）
        if txt_path.exists():
            print(f"跳过: {txt_path.name} 已存在")
            if not keep_audio and mp3_path.exists():
                safe_delete(mp3_path)
            continue

        try:
            # 转录
            text = transcribe_audio(mp3_path, model)
            if not text.strip():
                print(f"警告: {mp3_path.name} 转录结果为空，跳过删除")
                continue

            # 写入文本
            txt_path.write_text(text, encoding="utf-8")
            processed += 1
            print(f"已保存: {txt_path.name}")

            # 删除原始音频（仅当不保留时）
            if not keep_audio:
                safe_delete(mp3_path)

        except Exception as e:
            print(f"错误转录 {mp3_path.name}: {e}", file=sys.stderr)
            logger.error(f"处理失败: {mp3_path.name} - {e}")

    mode = "保留音频" if keep_audio else "已删除音频"
    print(f"全部完成！共处理 {processed} 个文件，{mode}。")


if __name__ == "__main__":
    main()

openai-whisper从多媒体抽取文本（windows GPU 版本）

drake — Wed, 12 Nov 2025 23:41:03 +0000

GPU 型号

NVIDIA GeForce GTX 750 Ti

依赖

python
三方包

pip uninstall torch torchvision torchaudio -y
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
pip install -U openai-whisper

echo "智能批量转录：自动选择最大可用 Whisper 模型"

# 方法1：尝试用 nvidia-smi 获取显存（兼容 Git Bash）
TOTAL_MEM=0
if command -v nvidia-smi > /dev/null 2>&1; then
    MEM_LINE=$(nvidia-smi --query-gpu=memory.total --format=csv,noheader,nounits | head -1)
    if [[ $MEM_LINE =~ ^[0-9]+$ ]]; then
        TOTAL_MEM=$MEM_LINE
    fi
fi

# 方法2：如果 nvidia-smi 失败，fallback 到 PyTorch 检测（推荐！）
if [ $TOTAL_MEM -eq 0 ] && command -v python > /dev/null 2>&1; then
    MEM_PY=$(python -c "
import torch
if torch.cuda.is_available():
    print(torch.cuda.get_device_properties(0).total_memory // 1024 // 1024)
else:
    print(0)
" 2>/dev/null || echo 0)
    TOTAL_MEM=$((MEM_PY))
fi

echo "检测到显存: ${TOTAL_MEM} MiB"

# 正确模型选择逻辑（显存越大，模型越大）
if [ $TOTAL_MEM -ge 20000 ]; then
    MODEL="large-v3"
elif [ $TOTAL_MEM -ge 14000 ]; then
    MODEL="medium"
elif [ $TOTAL_MEM -ge 8000 ]; then
    MODEL="small"
elif [ $TOTAL_MEM -ge 4000 ]; then
    MODEL="base"
else
    MODEL="tiny"
fi

echo "自动选择模型: $MODEL"
echo "=================================================="

# 检查 whisper 命令是否存在
if ! command -v whisper > /dev/null 2>&1; then
    echo "错误：whisper 命令未找到！请先 pip install openai-whisper"
    exit 1
fi

# 批量转录
count=0
for mp3 in *.mp3; do
    [[ -f "$mp3" ]] || continue
    txt="${mp3%.mp3}.txt"

    if [ ! -f "$txt" ]; then
        echo "转录: $mp3 → $txt (模型: $MODEL)"
        whisper "$mp3" \
            --model "$MODEL" \
            --device cuda \
            --output_format txt \
            --verbose False \
            --output_dir .
        ((count++))
    else
        echo "跳过: $txt 已存在"
    fi
done

echo "=================================================="
echo "全部完成！本次新增转录 $count 个文件。"

faster whisper从多媒体语音材料中抽取出文本-4

drake — Wed, 12 Nov 2025 08:41:43 +0000

"""
批量转录当前目录下的 .mp3 文件，使用 faster-whisper
转录完成后立即删除原始 .mp3 文件，仅保留 .txt
"""
import os
import sys
import logging
import argparse
from pathlib import Path
from faster_whisper import WhisperModel

# ================== 日志配置 ==================
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('Extract Text')

# ================== 配置区 ==================
MODEL_SIZE = "base"        # 可选: tiny, base, small, medium, large
DEVICE = "cpu"              # cpu 或 cuda
COMPUTE_TYPE = "int8"       # int8, float16, float32 (CPU 推荐 int8)
VAD_FILTER = True           # 启用语音活动检测，去除静音
OUTPUT_FORMAT = "txt"       # 输出格式
VERBOSE = True              # 是否显示详细日志

# ===========================================

def transcribe_audio(audio_path: Path, model: WhisperModel) -> str:
    """转录单个音频文件，返回文本内容"""
    print(f"转录: {audio_path.name} → {audio_path.stem}.txt")

    segments, info = model.transcribe(
        str(audio_path),
        language=None,
        beam_size=5,
        vad_filter=VAD_FILTER,
        vad_parameters=dict(min_silence_duration_ms=500),
        word_timestamps=False,
    )

    text_lines = []
    for segment in segments:
        line = segment.text.strip()
        if line:
            text_lines.append(line)
        if VERBOSE:
            logger.info(f"[{segment.start:06.2f}s --> {segment.end:06.2f}s] {line}")

    return "\n".join(text_lines)


def safe_delete(file_path: Path):
    """安全删除文件，捕获异常并记录"""
    try:
        file_path.unlink()
        logger.info(f"已删除: {file_path.name}")
    except Exception as e:
        logger.error(f"删除失败 {file_path.name}: {e}")


def main(keep_audio: bool = False):
    print("=== faster-whisper 批量转录 + 自动清理音频 ===")

    current_dir = Path(".")
    mp3_files = sorted(current_dir.glob("*.mp3"))

    if not mp3_files:
        print("未找到 .mp3 文件，退出。")
        return

    # 加载模型（只加载一次）
    print(f"正在加载模型 {MODEL_SIZE} ({DEVICE}, {COMPUTE_TYPE})...")
    model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE)

    processed = 0
    for mp3_path in mp3_files:
        txt_path = mp3_path.with_suffix(".txt")

        # 如果 .txt 已存在，跳过（但仍可选择删除 .mp3）
        if txt_path.exists():
            print(f"跳过: {txt_path.name} 已存在")
            if not keep_audio and mp3_path.exists():
                safe_delete(mp3_path)
            continue

        try:
            # 转录
            text = transcribe_audio(mp3_path, model)
            if not text.strip():
                print(f"警告: {mp3_path.name} 转录结果为空，跳过删除")
                continue

            # 写入文本
            txt_path.write_text(text, encoding="utf-8")
            processed += 1
            print(f"已保存: {txt_path.name}")

            # 删除原始音频（仅当不保留时）
            if not keep_audio:
                safe_delete(mp3_path)

        except Exception as e:
            print(f"错误转录 {mp3_path.name}: {e}", file=sys.stderr)
            logger.error(f"处理失败: {mp3_path.name} - {e}")

    mode = "保留音频" if keep_audio else "已删除音频"
    print(f"全部完成！共处理 {processed} 个文件，{mode}。")


if __name__ == "__main__":
    main()

faster whisper从多媒体语音材料中抽取出文本-3

drake — Wed, 12 Nov 2025 04:47:55 +0000

"""
批量转录当前目录下的 .mp3 文件，使用 faster-whisper
"""

import os
import logging
import sys
from pathlib import Path
from faster_whisper import WhisperModel

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('Extrat Text')

# ================== 配置区 ==================
MODEL_SIZE = "small"          # 可选: tiny, base, small, medium, large
DEVICE = "cpu"                # cpu 或 cuda
COMPUTE_TYPE = "int8"         # int8, float16, float32 (CPU 推荐 int8)
VAD_FILTER = True             # 启用语音活动检测，去除静音
OUTPUT_FORMAT = "txt"         # 只输出 .txt
VERBOSE = True               # 是否显示详细日志
# ===========================================

def transcribe_audio(audio_path: Path, model: WhisperModel) -> str:
    """转录单个音频文件，返回文本内容"""
    print(f"转录: {audio_path.name} → {audio_path.stem}.txt")

    segments, info = model.transcribe(
        str(audio_path),
        language=None,           # 自动检测
        beam_size=5,
        vad_filter=VAD_FILTER,
        vad_parameters=dict(min_silence_duration_ms=500),
        word_timestamps=False,
    )

    text_lines = []
    for segment in segments:
        line = segment.text.strip()
        text_lines.append(line)
        if VERBOSE:
            logger.info(".", end="", flush=True)
            # print(f"[{segment.start:06.2f}s --> {segment.end:06.2f}s] {line}", flush=True)

    return "\n".join(text_lines)


def main():
    print("=== faster-whisper 批量转录 ===")

    current_dir = Path(".")
    mp3_files = sorted(current_dir.glob("*.mp3"))

    if not mp3_files:
        print("未找到 .mp3 文件，退出。")
        return

    # 加载模型（只加载一次）
    print(f"正在加载模型 {MODEL_SIZE} ({DEVICE}, {COMPUTE_TYPE})...")
    model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE)

    processed = 0
    for mp3_path in mp3_files:
        txt_path = mp3_path.with_suffix(".txt")

        if txt_path.exists():
            print(f"跳过: {txt_path.name} 已存在")
            continue

        try:
            text = transcribe_audio(mp3_path, model)
            txt_path.write_text(text, encoding="utf-8")
            processed += 1
        except Exception as e:
            print(f"错误转录 {mp3_path.name}: {e}", file=sys.stderr)

    print(f"全部完成！共处理 {processed} 个文件。")


if __name__ == "__main__":
    main()

faster whisper从多媒体语音材料中抽取出文本-2

drake — Wed, 12 Nov 2025 04:43:04 +0000

为脚本添加每个音频的时长统计和每个音频转换所有的耗时统计

安装依赖

pip install faster-whisper pydub

"""
批量转录当前目录下的 .mp3 文件，使用 faster-whisper
新增功能：
- 每个音频的时长（秒）
- 每个音频的转录耗时（秒）
- 总计统计：总音频时长、总转录耗时、平均实时倍率
"""
import os
import sys
import time
from pathlib import Path
from typing import List, Tuple

from faster_whisper import WhisperModel
from pydub import AudioSegment


# ================== 配置区 ==================
MODEL_SIZE = "small"      # 可选: tiny, base, small, medium, large
DEVICE = "cpu"            # cpu 或 cuda
COMPUTE_TYPE = "int8"     # int8, float16, float32 (CPU 推荐 int8)
VAD_FILTER = True         # 启用语音活动检测，去除静音
OUTPUT_FORMAT = "txt"     # 只输出 .txt
VERBOSE = True            # 是否显示详细日志
# ===========================================


def get_audio_duration(audio_path: Path) -> float:
    """使用 pydub 获取音频时长（秒）"""
    try:
        audio = AudioSegment.from_file(str(audio_path))
        return len(audio) / 1000.0  # 毫秒 → 秒
    except Exception as e:
        print(f"无法获取 {audio_path.name} 时长: {e}", file=sys.stderr)
        return 0.0


def transcribe_audio(
    audio_path: Path, model: WhisperModel
) -> Tuple[str, float, float]:
    """
    转录单个音频文件
    返回: (文本内容, 音频时长秒, 转录耗时秒)
    """
    duration = get_audio_duration(audio_path)
    print(f"转录: {audio_path.name} ({duration:.2f}s) → {audio_path.stem}.txt")

    start_time = time.perf_counter()
    segments, info = model.transcribe(
        str(audio_path),
        language=None,           # 自动检测
        beam_size=5,
        vad_filter=VAD_FILTER,
        vad_parameters=dict(min_silence_duration_ms=500),
        word_timestamps=False,
    )
    elapsed = time.perf_counter() - start_time

    text_lines = []
    for segment in segments:
        line = segment.text.strip()
        text_lines.append(line)
        if VERBOSE:
            print(f"[{segment.start:06.2f}s --> {segment.end:06.2f}s] {line}", flush=True)

    return "\n".join(text_lines), duration, elapsed


def format_time(seconds: float) -> str:
    """将秒数格式化为 h:mm:ss"""
    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = seconds % 60
    return f"{hours}:{minutes:02d}:{secs:05.2f}"


def main():
    print("=== faster-whisper 批量转录（带时长与耗时统计）===")

    current_dir = Path(".")
    mp3_files = sorted(current_dir.glob("*.mp3"))

    if not mp3_files:
        print("未找到 .mp3 文件，退出。")
        return

    # 加载模型（只加载一次）
    print(f"正在加载模型 {MODEL_SIZE} ({DEVICE}, {COMPUTE_TYPE})...")
    model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE)

    processed = 0
    total_audio_duration = 0.0
    total_transcribe_time = 0.0
    results = []

    for mp3_path in mp3_files:
        txt_path = mp3_path.with_suffix(".txt")
        if txt_path.exists():
            duration = get_audio_duration(mp3_path)
            print(f"跳过: {txt_path.name} 已存在 ({duration:.2f}s)")
            total_audio_duration += duration
            continue

        try:
            text, duration, elapsed = transcribe_audio(mp3_path, model)
            txt_path.write_text(text, encoding="utf-8")

            total_audio_duration += duration
            total_transcribe_time += elapsed
            processed += 1

            rtf = elapsed / duration if duration > 0 else float('inf')
            print(f"完成: {mp3_path.name} | 时长 {duration:.2f}s | 耗时 {elapsed:.2f}s | RTF {rtf:.2f}x")
            results.append((mp3_path.name, duration, elapsed, rtf))

        except Exception as e:
            print(f"错误转录 {mp3_path.name}: {e}", file=sys.stderr)

    # ================== 汇总统计 ==================
    print("\n" + "=" * 60)
    print("转录完成汇总")
    print("=" * 60)
    print(f"成功处理文件数   : {processed}")
    print(f"总音频时长       : {format_time(total_audio_duration)}")
    print(f"总转录耗时       : {format_time(total_transcribe_time)}")
    if total_audio_duration > 0:
        avg_rtf = total_transcribe_time / total_audio_duration
        print(f"平均实时倍率(RTF): {avg_rtf:.2f}x")
    else:
        print(f"平均实时倍率(RTF): N/A")

    if results:
        print(f"\n明细列表：")
        print(f"{'文件名':<40} {'音频时长':>10} {'转录耗时':>10} {'RTF':>8}")
        print("-" * 70)
        for name, dur, ela, rtf in results:
            print(f"{name:<40} {dur:10.2f}s {ela:10.2f}s {rtf:8.2f}x")

    print("=" * 60)


if __name__ == "__main__":
    main()

faster whisper从多媒体语音材料中抽取出文本

drake — Wed, 12 Nov 2025 04:38:45 +0000

"""
批量转录当前目录下的 .mp3 文件，使用 faster-whisper
"""

import os
import sys
from pathlib import Path
from faster_whisper import WhisperModel

# ================== 配置区 ==================
MODEL_SIZE = "small"          # 可选: tiny, base, small, medium, large
DEVICE = "cpu"                # cpu 或 cuda
COMPUTE_TYPE = "int8"         # int8, float16, float32 (CPU 推荐 int8)
VAD_FILTER = True             # 启用语音活动检测，去除静音
OUTPUT_FORMAT = "txt"         # 只输出 .txt
VERBOSE = True               # 是否显示详细日志
# ===========================================

def transcribe_audio(audio_path: Path, model: WhisperModel) -> str:
    """转录单个音频文件，返回文本内容"""
    print(f"转录: {audio_path.name} → {audio_path.stem}.txt")

    segments, info = model.transcribe(
        str(audio_path),
        language=None,           # 自动检测
        beam_size=5,
        vad_filter=VAD_FILTER,
        vad_parameters=dict(min_silence_duration_ms=500),
        word_timestamps=False,
    )

    text_lines = []
    for segment in segments:
        line = segment.text.strip()
        text_lines.append(line)
        if VERBOSE:
            print(".", end="", flush=True)
            # print(f"[{segment.start:06.2f}s --> {segment.end:06.2f}s] {line}", flush=True)

    return "\n".join(text_lines)


def main():
    print("=== faster-whisper 批量转录 ===")

    current_dir = Path(".")
    mp3_files = sorted(current_dir.glob("*.mp3"))

    if not mp3_files:
        print("未找到 .mp3 文件，退出。")
        return

    # 加载模型（只加载一次）
    print(f"正在加载模型 {MODEL_SIZE} ({DEVICE}, {COMPUTE_TYPE})...")
    model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE)

    processed = 0
    for mp3_path in mp3_files:
        txt_path = mp3_path.with_suffix(".txt")

        if txt_path.exists():
            print(f"跳过: {txt_path.name} 已存在")
            continue

        try:
            text = transcribe_audio(mp3_path, model)
            txt_path.write_text(text, encoding="utf-8")
            processed += 1
        except Exception as e:
            print(f"错误转录 {mp3_path.name}: {e}", file=sys.stderr)

    print(f"全部完成！共处理 {processed} 个文件。")


if __name__ == "__main__":
    main()

python 连接 bittensor的子网

drake — Mon, 27 Oct 2025 13:18:01 +0000

安装依赖

# 使用 pip 安装
pip install bittensor

# 或安装最新开发版
pip install git+https://github.com/opentensor/bittensor.git

代码

import bittensor as bt

# 创建子网连接
subtensor = bt.subtensor(network='finney')  # 主网

# 查询 Subnet 55 信息
metagraph = subtensor.metagraph(netuid=55)

print(f"Subnet 55 有 {len(metagraph.neurons)} 个神经元")