准备工作：环境准备
1.一个Google账号
2.一张能使用的国际信用卡或借记卡（visa/万事达/JCB等等）
3.一个telegram账号
开始：创建实例
打开Google云网址：https://console.cloud.google.com/
2.使用Google账号登录
3.点击导航栏
4.点击compute engine-虚拟机实例
5.创建实例
名称可以不用管，区域选俄勒冈，可用区不用管
选E2，日常计算费用低，机器类型选共享核心，e2-micro
操作系统和储存空间，点击中间的更改按钮
操作系统选择Ubuntu（乌班图）
其他不用管
数据保护一定要选择无备份（免费的关键）
防火墙选择允许HTTP流量,允许HTTPS流量，其余不用管
选择完后，直接点击创建即可之后还需要创建防火墙规则，也是非常简单跟着操作一步一步来即可
创建一条入站防火墙规则：
创建一条出站防火墙规则：
经过以上的所有步骤我们就已经成功把我们自己免费的的VPS服务器创建好了下面跟着我的步骤一步一步部署我们的7*24小时盯盘系统
点击我们刚刚创建的虚拟机实例的SSH系统会跳出一个授权窗口，点击确定即可
第一步：系统底层组件初始化 命令执行
由于 minimal 版本的 Ubuntu 非常精简，很多基础工具被阉割了。首先需要更新软件源，并补齐 Python3 环境和文本编辑器（nano）：
Bash
sudo apt update && sudo apt install -y python3 python3-pip nano第二步：注入量化与 AI 核心依赖
接下来是安装 Python 的第三方库。涵盖了交易所直连（ccxt）、数据处理（pandas）、大模型风控以及 Telegram 预警模块、requests网络网络请求库：
Bash
pip3 install ccxt pandas google-generativeai python-telegram-bot requests如果以上代码有报错，则运行以下替代代码：
Bash
pip3 install ccxt pandas google-genai python-telegram-bot requests第三步：直接开始编写代码 核心步骤
（你可以找ai给你写一个你习惯的盯盘策略，代码直接让ai给你写成Python代码即可）由于我已经有了自己的代码，我这里直接展示一下具体操作
在SSH窗口中输入：nano strategy_engine.py这个.py文件名是你自己取的，写什么都可以的
通常你打开这个.py文件是空白的，需要粘贴上你找ai给你编写的py盯盘代码，这里我就已经贴上我的MA99盯盘代码（如有需要可以去我的博客获取）
另外你需要准备三个API密钥：
1.Gemini API KEY（可以去Google studio获取/其他大模型也同理，但是代码中调佣的大模型也要改）
2.你个人的telegram BOT TOKEN
3.你个人的telegram CHAT ID
第二部分：如何让它在你关掉电脑后继续运行？（极为关键）重点
如果你像以前一样直接输入 python3 strategy_engine.py，那么一旦你关闭了这个谷歌云的网页黑框，程序就会立刻死掉。
为了让它真正运行在云端后台，你需要使用 nohup（no hang up，不挂断）命令。
在终端里输入这行充满魔法的命令并回车：
Bash
nohup python3 strategy_engine.py > bot.log 2>&1 &解析： 这句话的意思是“把这个程序丢到后台去悄悄跑，不要受我关网页的影响，并且把所有原本要打印在屏幕上的字，都偷偷记到一个叫 bot.log 的日记本文件里。”
敲完回车后，它会弹出一个进程号（比如 [1] 12345），这就说明永动机已经成功点火了！ 你现在就算把电脑砸了，它依然会在美国的谷歌机房里 24 小时替你盯盘。
如果你想看看它在后台干嘛，只需要输入：
Bash
tail -f bot.log这会实时滚动显示它的日记（按 Ctrl + C 可以退出查看，不影响后台运行）。
最后展示手机上的telegram的提醒播报：
写在最后，这个教程其实已经极其简单了，只要跟着一步一步操作，或者有一些云服务器部署的经验甚至都可以自己手搓出来，我把这个教程整理出来目的就是让更多人了解这个可以不用自己盯盘，一些策略可以用工具来替代我们人力监控（甚至可以在代码上接入自动交易，这个如果有需要我后面也可以更新出来），如果本期教程能帮到你，请务必给我点个订阅和点赞，十分需要你的大力支持！
实战演练：MA 99 策略 V8 自动化引擎源码
以下是我前面使用的python脚本，综合实际结合了多时间级别 (MTF) 共振审查与 AI 辅助分析的完整 Python 自动化交易脚本，有需要可以自行复制使用：
import ccxt
import pandas as pd
import requests
import time
import json
import os
import logging
from datetime import datetime
from google import genai
# ================= 1. 专业日志配置 =================
logging.basicConfig(level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
logger = logging.getLogger(__name__)
# ================= 2. 核心参数与 API 设置 =================
SYMBOLS = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'BNB/USDT', 'XRP/USDT', 'ADA/USDT', 'AVAX/USDT', 'LINK/USDT', 'DOGE/USDT', 'DOT/USDT']
TIMEFRAMES = ['15m', '1h', '4h']
MA_LENGTH = 99
LOOKBACK_BARS = 15
# 【V8 新增】大周期审查映射表
HTF_MAP = {
'15m': ['1h', '4h'],
'1h': ['4h'],
'4h': ['1d']
}
# 🔑 你的三大核心密钥（你需要对应去获取）
GEMINI_API_KEY = "你的GEMINI_API_KEY"
TELEGRAM_BOT_TOKEN = "你的TELEGRAM_BOT_TOKEN"
TELEGRAM_CHAT_ID = "你的TELEGRAM_CHAT_ID"
MEMORY_FILE = "signal_memory.json"
client = genai.Client(api_key=GEMINI_API_KEY)
exchange = ccxt.binanceus({'enableRateLimit': True})
# ================= 3. 记忆与 AI 模块 =================
def load_memory():
if os.path.exists(MEMORY_FILE):
try:
with open(MEMORY_FILE, 'r') as f: return json.load(f)
except: pass
return {}
def save_memory(memory_dict):
try:
with open(MEMORY_FILE, 'w') as f: json.dump(memory_dict, f)
except: pass
def get_ai_analysis(symbol, timeframe, action, price, rsi):
prompt = f"""
你现在是一位拥有10年经验的量化交易员。当前时间：{datetime.now().strftime('%Y-%m-%d')}。
交易标的：{symbol} ({timeframe} 级别)，当前价格：{price}
技术面：当前触发【{action}】信号，且已通过大周期(MTF)的顺势共振审查！RSI为 {rsi:.1f}。
请给出100字以内的极简操作建议、止损位和盈亏比参考。客观理性，直接给结论。
"""
try:
response = client.models.generate_content(model='gemini-2.5-flash', contents=prompt)
return response.text.strip()
except:
return "⚠️ AI 分析暂时不可用，请严格带好止损独立决策。"
def send_telegram_alert(symbol, timeframe, action, price, rsi, ai_advice):
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
action_str = "🟢 做多 (LONG)" if action == 'LONG' else "🔴 做空 (SHORT)"
text = f"🚨 <b>MTF 共振狙击预警</b> 🚨\n\n" \
f"📌 <b>标的:</b> {symbol} | <b>信号周期:</b> {timeframe}\n" \
f"🎯 <b>方向:</b> {action_str} (已确认大趋势顺向)\n" \
f"💰 <b>现价:</b> ${price}\n" \
f"📊 <b>动能:</b> RSI = {rsi:.1f}\n" \
f"⏰ <b>时间:</b> {datetime.now().strftime('%H:%M:%S')}\n\n" \
f"🧠 <b>AI 战术板:</b>\n<i>{ai_advice}</i>"
try:
requests.post(url, json={"chat_id": TELEGRAM_CHAT_ID, "text": text, "parse_mode": "HTML"}, timeout=10)
except Exception as e:
logger.error(f"Telegram 推送失败: {e}")
# ================= 4. 【V8 新增】大周期审查官 =================
def verify_htf_alignment(symbol, timeframe, action):
htfs = HTF_MAP.get(timeframe, [])
if not htfs: return True
for htf in htfs:
try:
bars = exchange.fetch_ohlcv(symbol, timeframe=htf, limit=120)
if not bars or len(bars) < 100: return False
df = pd.DataFrame(bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df = df.iloc[:-1] # 抛弃未走完的 K 线
df['MA99'] = df['close'].rolling(window=MA_LENGTH).mean()
current_htf_close = df.iloc[-1]['close']
current_htf_ma99 = df.iloc[-1]['MA99']
# 【一票否决制】
if action == "LONG" and current_htf_close < current_htf_ma99:
logger.info(f"🚫 拦截伪信号: {symbol} {timeframe} 做多，但大周期 {htf} 为空头趋势 (被 MA99 压制)。")
return False
if action == "SHORT" and current_htf_close > current_htf_ma99:
logger.info(f"🚫 拦截伪信号: {symbol} {timeframe} 做空，但大周期 {htf} 为多头趋势 (受 MA99 支撑)。")
return False
except Exception as e:
logger.error(f"大周期 {htf} 获取异常: {e}")
return False # 出错时安全第一，直接拦截
return True # 所有大周期全部通过审查！
# ================= 5. 策略核心引擎 =================
def check_ma99_strategy(symbol, timeframe):
try:
bars = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=200)
if not bars or len(bars) < 150: return None, None, None
df = pd.DataFrame(bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df = df.iloc[:-1].copy()
df['MA25'] = df['close'].rolling(window=25).mean()
df['MA99'] = df['close'].rolling(window=MA_LENGTH).mean()
high_low = df['high'] - df['low']
high_close = (df['high'] - df['close'].shift()).abs()
low_close = (df['low'] - df['close'].shift()).abs()
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
df['ATR'] = tr.rolling(14).mean()
delta = df['close'].diff()
gain = delta.where(delta > 0, 0.0).ewm(alpha=1/14, adjust=False).mean()
loss = (-delta.where(delta < 0, 0.0)).ewm(alpha=1/14, adjust=False).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
current = df.iloc[-1]
is_bull_trend = current['MA25'] > current['MA99']
long_pullback = (current['low'] <= current['MA99'] + (0.5 * current['ATR'])) and (current['close'] > current['MA99'])
long_rsi_ok = 40 < current['RSI'] < 65
if is_bull_trend and long_pullback and long_rsi_ok:
return "LONG", current, current['RSI']
is_bear_trend = current['MA25'] < current['MA99']
short_pullback = (current['high'] >= current['MA99'] - (0.5 * current['ATR'])) and (current['close'] < current['MA99'])
short_rsi_ok = 35 < current['RSI'] < 60
if is_bear_trend and short_pullback and short_rsi_ok:
return "SHORT", current, current['RSI']
return None, current, None
except Exception as e:
return None, None, None
# ================= 6. 守护进程中枢 =================
if __name__ == "__main__":
import warnings
warnings.filterwarnings('ignore')
logger.info(f"🚀 V8 引擎 (含大周期上帝视角) 启动！")
signal_memory = load_memory()
while True:
try:
for symbol in SYMBOLS:
for tf in TIMEFRAMES:
signal, last_bar, rsi = check_ma99_strategy(symbol, tf)
if not signal or last_bar is None: continue
# 【V8 新增】：查大周期底牌！如果大周期不配合，直接抛弃该信号
if not verify_htf_alignment(symbol, tf, signal):
continue
memory_key = f"{symbol}_{tf}_{signal}"
current_bar_time = last_bar['timestamp']
if signal_memory.get(memory_key) == current_bar_time: continue
signal_memory[memory_key] = current_bar_time
save_memory(signal_memory)
logger.info(f"🎯 黄金共振信号确认！{symbol} ({tf}) -> {signal}，现价 {last_bar['close']}")
ai_advice = get_ai_analysis(symbol, tf, signal, last_bar['close'], rsi)
send_telegram_alert(symbol, tf, signal, last_bar['close'], rsi, ai_advice)
time.sleep(60)
except Exception as e:
time.sleep(60)
🚀 觉得教程有用的话？一定别忘了订阅我的 YouTube 频道获取更多黑科技！
❤️ 一键订阅 TECHTRADER HUi