# -*- coding: utf-8 -*-
import os
import sys
import yaml
import argparse
import requests
import pymysql.cursors
import random
import re
import datetime
from openai import OpenAI
from dotenv import load_dotenv
from PIL import Image, ImageDraw, ImageFont
# ==============================================================================
# 1. CONFIGURATION & ENVIRONNEMENT
# ==============================================================================
def log(msg):
print(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(SCRIPT_DIR, '.env'))
def get_env(key, default=None):
return os.getenv(key, default)
parser = argparse.ArgumentParser()
parser.add_argument('--wp-path', type=str, required=True, help='Root path of WP')
parser.add_argument('--action_type', type=str, default='feed_morning', help='Action to perform')
parser.add_argument('--status', type=str, default='publish', choices=['publish', 'draft'])
args, unknown = parser.parse_known_args()
log(f"🚀 Booting Master Bot (Action: {args.action_type} | Status: {args.status})...")
# --- LECTURE DIRECTE DEPUIS .ENV ---
BOT_SECRET = get_env('BOT_SECRET_KEY')
BOT_AUTHOR_ID = int(get_env('BOT_AUTHOR_ID', 1))
if not BOT_SECRET:
log("CRITICAL FAILURE: BOT_SECRET_KEY missing in .env")
sys.exit(1)
SITE_DOMAIN = get_env('WP_SITE_DOMAIN', 'https://gapandhold.com').rstrip('/')
LLM_MODEL = get_env('LLM_MODEL', 'gpt-4o')
WP_REST_FEED = f"{SITE_DOMAIN}/wp-json/etea/v1/bot-activity/"
WP_REST_INTERACT = f"{SITE_DOMAIN}/wp-json/etea/v1/bot-interact/"
openai_api_key = get_env('LLM_API_KEY')
if not openai_api_key:
log("CRITICAL FAILURE: openai_api_key missing in .env")
sys.exit(1)
client = OpenAI(
api_key=openai_api_key,
base_url=get_env('LLM_BASE_URL')
)
# ==============================================================================
# 2. AUTO-DÉTECTION DB SÉCURISÉE (pymysql)
# ==============================================================================
def get_wp_config_creds(wp_path):
creds = {}
current = wp_path
while len(current) > 4:
cfg = os.path.join(current, 'wp-config.php')
if os.path.exists(cfg):
with open(cfg, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
for k in ['DB_NAME', 'DB_USER', 'DB_PASSWORD', 'DB_HOST']:
m = re.search(rf"define\(\s*['\"]{k}['\"]\s*,\s*['\"](.*?)['\"]\s*\);", content)
if m: creds[k] = m.group(1)
m_pref = re.search(r"\$table_prefix\s*=\s*['\"](.*?)['\"];", content)
creds['table_prefix'] = m_pref.group(1) if m_pref else 'wp_'
return creds
current = os.path.dirname(current)
return None
wp_creds = get_wp_config_creds(args.wp_path)
if not wp_creds:
log("CRITICAL FAILURE: wp-config.php not found.")
sys.exit(1)
DB_PREFIX = wp_creds.get('table_prefix', 'wp_')
def get_db_connection():
try:
return pymysql.connect(
host=wp_creds.get('DB_HOST'), user=wp_creds.get('DB_USER'),
password=wp_creds.get('DB_PASSWORD'), database=wp_creds.get('DB_NAME'),
cursorclass=pymysql.cursors.DictCursor
)
except Exception as e:
log(f"CRITICAL FAILURE: DB Connection failed: {e}")
sys.exit(1)
db = get_db_connection()
try:
with db.cursor() as cursor:
# ==============================================================================
# 3. ROUTAGE DES ACTIONS
# ==============================================================================
# ---------------------------------------------------------
# ACTION A : VOTER SUR UN TICKER
# ---------------------------------------------------------
if args.action_type == 'vote':
log("Executing Vote logic...")
cursor.execute(f"""
SELECT ticker, direction_anticipee
FROM {DB_PREFIX}etea_earnings_analysis
WHERE earnings_date >= CURDATE() AND direction_anticipee != 'Neutral'
ORDER BY RAND() LIMIT 1
""")
s = cursor.fetchone()
if not s:
log("No signal available for voting.")
sys.exit(0)
vote_val = 'agree' if random.random() > 0.1 else 'disagree'
url = f"{WP_REST_INTERACT}?bot_key={BOT_SECRET}&action_type=vote&author_id={BOT_AUTHOR_ID}&target={s['ticker']}&vote_value={vote_val}"
try:
res = requests.post(url, verify=False)
log(f"Voted '{vote_val}' on ${s['ticker']}. WP Response: {res.text}")
except Exception as e:
log(f"Vote Failed: {e}")
sys.exit(0)
# ---------------------------------------------------------
# ACTION B : COMMENTER SUR LA PAGE D'UN TICKER
# ---------------------------------------------------------
elif args.action_type == 'ticker_comment':
log("Executing Ticker Comment logic...")
cursor.execute(f"""
SELECT ticker, direction_anticipee, confidence_score, anticipated_post_gap_strategy
FROM {DB_PREFIX}etea_earnings_analysis
WHERE earnings_date >= CURDATE() AND direction_anticipee != 'Neutral'
ORDER BY RAND() LIMIT 1
""")
s = cursor.fetchone()
if not s:
log("No signal available to comment on.")
sys.exit(0)
prompt = f"Write a professional 2-sentence quantitative trader comment about ${s['ticker']}. The AI model indicates a {s['direction_anticipee']} bias (Confidence: {float(s['confidence_score'])*100:.1f}%). Strategy to execute: {s['anticipated_post_gap_strategy']}. Do NOT use markdown (*). Use plain text."
try:
comp = client.chat.completions.create(model=LLM_MODEL, messages=[{"role":"user", "content": prompt}], temperature=0.7)
comment_text = re.sub(r'.*?', '', comp.choices[0].message.content, flags=re.DOTALL).strip()
comment_text = comment_text.replace('**', '')
url = f"{WP_REST_INTERACT}?bot_key={BOT_SECRET}&action_type=ticker_comment&author_id={BOT_AUTHOR_ID}&target={s['ticker']}"
res = requests.post(url, json={'content': comment_text}, verify=False)
log(f"Commented on ${s['ticker']}. WP Response: {res.text}")
except Exception as e:
log(f"Comment Failed: {e}")
sys.exit(0)
# ---------------------------------------------------------
# ACTION C : CRÉER UN POST DANS LE GLOBAL ACTIVITY FEED
# ---------------------------------------------------------
else:
log(f"Executing Global Feed Post logic (Type: {args.action_type})...")
is_evening = 'evening' in args.action_type
query = f"""
SELECT T1.ticker, T1.security, T1.direction_anticipee, T1.confidence_score,
T1.earnings_date, T1.earnings_timing, T1.actual_gap_percent,
T1.gap_anticipation_correct, T2.logo_url, T2.security as full_name
FROM {DB_PREFIX}etea_earnings_analysis T1
LEFT JOIN {DB_PREFIX}etea_tickers_list T2 ON T1.ticker = T2.ticker
"""
if is_evening:
query += " WHERE T1.earnings_date = CURDATE() AND T1.actual_gap_percent IS NOT NULL ORDER BY ABS(T1.actual_gap_percent) DESC LIMIT 3"
else:
query += " WHERE T1.earnings_date >= CURDATE() AND T1.direction_anticipee IN ('Bullish', 'Bearish') ORDER BY T1.earnings_date ASC, T1.confidence_score DESC LIMIT 3"
cursor.execute(query)
signals = cursor.fetchall()
if not signals:
log("⚠️ No signals found for this time period. Aborting post.")
sys.exit(0)
# 1. PRÉPARATION DU TEXTE
ctx_lines = []
for s in signals:
if is_evening:
res = "WIN" if s['gap_anticipation_correct'] else "LOSS"
ctx_lines.append(f"- ${s['ticker']}: Actual Gap {s['actual_gap_percent']}% ({res})")
else:
ctx_lines.append(f"- ${s['ticker']}: {s['direction_anticipee']} (Conf: {float(s['confidence_score'])*100:.1f}%)")
context_text = "\n".join(ctx_lines)
prompts = {
'feed_morning': f"Write a brief, punchy morning pre-market update for a trading community feed. Introduce the 'Gap & Hold' strategy context. Highlight these upcoming AI earnings signals using an HTML list (- ):\n{context_text}\nUse emojis, keep it professional but hype. Do not use Markdown (**).",
'feed_noon': f"Write a mid-day market check-in for traders. Remind them to check their watchlists and prep for these upcoming Gap & Hold earnings setups using an HTML list (
- ):\n{context_text}\nUse emojis, keep it analytical but engaging. Do not use Markdown.",
'feed_evening': f"Write an evening market wrap-up. Review today's action and prep the community for tomorrow's Gap & Hold setups using these AI signals using an HTML list (
- ):\n{context_text}\nUse emojis, emphasize trading discipline and data. Do not use Markdown.",
'feed_weekly': f"Write a weekly strategy recap for the community. Focus on preparing the portfolio for next week's Gap & Hold setups using these high-probability AI signals using an HTML list (
- ):\n{context_text}\nUse emojis, emphasize discipline and data. Do not use Markdown."
}
prompt = prompts.get(args.action_type, prompts['feed_morning'])
log(f"Generating insight text via {LLM_MODEL}...")
try:
comp = client.chat.completions.create(
model=LLM_MODEL,
messages=[
{"role":"system", "content": "You are the AI System Core of Stock Error 404, a quantitative trading platform. Format output cleanly for a web feed in HTML."},
{"role":"user", "content": prompt}
],
temperature=0.7
)
post_content = re.sub(r'.*?', '', comp.choices[0].message.content, flags=re.DOTALL).strip()
post_content = post_content.replace('**', '')
log("Text generation OK.")
except Exception as e:
log(f"CRITICAL FAILURE: Text generation failed: {e}")
sys.exit(1)
# 2. GÉNÉRATION IMAGE FALLBACK (PILLOW)
def generate_quant_card_fallback(signals_data):
log("🎨 Generating Ultra-Sharp Public Dashboard...")
# 1. DOUBLE RÉSOLUTION POUR LA NETTETÉ (2K)
W, H = 2000, 1200
img = Image.new('RGB', (W, H), color='#050508')
draw = ImageDraw.Draw(img)
# Tentative de chargement d'une police système (plus nette)
try:
# Chemins courants sur Linux/Ubuntu
font_bold = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 60)
font_main = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 45)
font_small = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 35)
except:
font_bold = font_main = font_small = ImageFont.load_default()
# 2. GRID BACKGROUND (Subtile)
for x in range(0, W, 80): draw.line([(x, 0), (x, H)], fill='#0f0f1a', width=2)
for y in range(0, H, 80): draw.line([(0, y), (W, y)], fill='#0f0f1a', width=2)
# 3. AJOUT DU LOGO DU SITE ET NOM (BRANDING)
try:
# Chemin vers votre logo Gap & Hold (Ă adapter)
site_logo_path = os.path.join(args.wp_path, 'wp-content/uploads/2026/04/fav-icon-gapandhold.svg')
if os.path.exists(site_logo_path):
site_logo = Image.open(site_logo_path).convert("RGBA").resize((120, 120))
img.paste(site_logo, (80, 50), mask=site_logo)
except: pass
draw.text((220, 75), "GAP & HOLD", fill="#ffffff", font=font_bold)
draw.text((220, 140), "QUANTITATIVE INTELLIGENCE TERMINAL", fill="#44475a", font=font_small)
# 4. GÉNÉRATION DES CARTES
y_offset = 250
for s in signals_data:
ticker = s['ticker']
is_bull = s['direction_anticipee'] == 'Bullish'
color = '#00ffa3' if is_bull else '#ff2e63'
# Carte avec effet de profondeur
draw.rectangle([80, y_offset, W-80, y_offset+260], fill='#0a0a12', outline='#1c1c2d', width=3)
draw.rectangle([80, y_offset, 100, y_offset+260], fill=color) # Barre d'Ă©tat
# Ticker et Nom (Net)
draw.text((260, y_offset+50), f"${ticker}", fill=color, font=font_bold)
sec_name = (s.get('full_name') or s.get('security') or ticker).upper()
draw.text((260, y_offset+130), f"{sec_name[:40]}", fill="#888eb2", font=font_main)
# Données Masquées (Protection Premium)
if is_evening:
status, label = "COMPLETED", "HIGH VOLATILITY" if abs(float(s['actual_gap_percent'] or 0)) > 4 else "VOLATILITY: NORMAL"
else:
status, label = f"SIGNAL: {s['direction_anticipee'].upper()}", "STRENGTH: HIGH CONVICTION" if float(s['confidence_score'] or 0) > 0.7 else "STRENGTH: QUALIFIED"
draw.text((1000, y_offset+70), status, fill="white", font=font_main)
draw.text((1000, y_offset+140), label, fill=color, font=font_main)
# Logo de l'Action (Rond et Net)
if s['logo_url']:
try:
logo_path = os.path.join(args.wp_path, s['logo_url'].lstrip('/'))
if os.path.exists(logo_path):
asset_logo = Image.open(logo_path).convert("RGBA").resize((130, 130))
# Masque anti-aliasing pour le cercle
mask = Image.new('L', (130, 130), 0)
ImageDraw.Draw(mask).ellipse((0, 0, 130, 130), fill=255)
img.paste(asset_logo, (115, y_offset+65), mask=mask)
except: pass
y_offset += 300
# 5. FILIGRANE DE SÉCURITÉ ET FOOTER
draw.text((W//2 - 300, H - 80), "© GAPANDHOLD.COM - AI RESEARCH SYSTEM", fill="#232336", font=font_small)
# Exportation Haute Qualité
output_filename = f"terminal-insight-{random.randint(1000,9999)}.png"
save_dir = os.path.join(args.wp_path, 'wp-content/uploads')
os.makedirs(save_dir, exist_ok=True)
save_path = os.path.join(save_dir, output_filename)
# Option 'optimize' et 'quality' pour la netteté
img.save(save_path, "PNG", optimize=True)
return f"{SITE_DOMAIN}/wp-content/uploads/{output_filename}"
# 3. GÉNÉRATION IMAGE PRINCIPALE (DALL-E 3)
log("Generating visual...")
image_url = ""
try:
if get_env('LLM_BASE_URL'):
raise Exception("Local LLM cannot generate DALL-E images.")
response = client.images.generate(
model="dall-e-3",
prompt=f"Cyberpunk financial dashboard showing glowing charts for stock symbols {signals[0]['ticker']}, {signals[1]['ticker'] if len(signals)>1 else ''}. High-tech aesthetic.",
size="1024x1024", n=1
)
if response and response.data and len(response.data) > 0:
image_url = response.data[0].url
log("Image generated via DALL-E.")
else:
raise Exception("Empty response from DALL-E")
except Exception as e:
log(f"DALL-E Failed ({e}), switching to Pillow fallback...")
image_url = generate_quant_card_fallback(signals)
# 4. PUBLICATION API
log(f"Dispatching to WP Feed (Status: {args.status})...")
target_url = f"{WP_REST_FEED}?bot_key={BOT_SECRET}&status={args.status}"
payload = {
'author_id': BOT_AUTHOR_ID,
'content': post_content,
'image_url': image_url
}
try:
res = requests.post(target_url, json=payload, headers={'Content-Type': 'application/json'}, timeout=45, verify=False)
if res.status_code == 200:
log(f"âś… SUCCESS! WP Response: {res.json()}")
else:
log(f"❌ CRITICAL FAILURE: WP API Error {res.status_code}: {res.text}")
except Exception as e:
log(f"CRITICAL FAILURE: Request to WP failed: {e}")
finally:
if db.open:
db.close()
