k
248 lines
9.1 KiB
Python
248 lines
9.1 KiB
Python
"""
|
|
Trading decision loop.
|
|
Called by the Truth Social scraper after each new post is analyzed and saved.
|
|
Iterates all active subscribers and executes trades on their behalf.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Dict
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import AsyncSessionLocal
|
|
from app.models import BotTrade, Post, Subscription
|
|
from app.services.crypto import decrypt_api_key
|
|
from app.services.hyperliquid import HyperliquidTrader
|
|
from app.services.price_store import price_store # noqa: F401 (used elsewhere)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Platform-wide thresholds (per-user values in Subscription override where applicable)
|
|
GLOBAL_MIN_CONFIDENCE = 80 # hard floor — user min_confidence must be >= this
|
|
MAX_HOLD_SECONDS = 3600 # force-close after 1 hour (shared across users)
|
|
|
|
|
|
# Per-trade locks prevent TP/SL and max-hold tasks from both calling close_and_finalize
|
|
# simultaneously on the same trade. Lock is process-local — with the atomic
|
|
# conditional UPDATE below, multi-process is still safe (losers become no-ops).
|
|
_close_locks: Dict[int, asyncio.Lock] = {}
|
|
|
|
|
|
def _lock_for(trade_id: int) -> asyncio.Lock:
|
|
lock = _close_locks.get(trade_id)
|
|
if lock is None:
|
|
lock = asyncio.Lock()
|
|
_close_locks[trade_id] = lock
|
|
return lock
|
|
|
|
|
|
async def process_post(post: Post, db: AsyncSession) -> None:
|
|
"""
|
|
Entry point called by the scraper after a new post is saved.
|
|
`db` is used only to fetch the subscriber list; each subscriber gets its own session.
|
|
"""
|
|
if not post.relevant:
|
|
return
|
|
if (post.ai_confidence or 0) < GLOBAL_MIN_CONFIDENCE:
|
|
logger.info("Post %d skipped: confidence %d < %d", post.id, post.ai_confidence, GLOBAL_MIN_CONFIDENCE)
|
|
return
|
|
if post.signal not in ('buy', 'short'):
|
|
logger.info("Post %d skipped: signal=%s is not actionable", post.id, post.signal)
|
|
return
|
|
|
|
asset = post.price_impact_asset or 'BTC'
|
|
side = 'long' if post.signal == 'buy' else 'short'
|
|
|
|
logger.info("Signal: post=%d asset=%s side=%s confidence=%d", post.id, asset, side, post.ai_confidence)
|
|
|
|
result = await db.execute(select(Subscription).where(Subscription.active == True)) # noqa: E712
|
|
subscribers = result.scalars().all()
|
|
|
|
if not subscribers:
|
|
logger.info("No active subscribers, skipping trade execution")
|
|
return
|
|
|
|
# Snapshot primitive fields so tasks don't share the ORM object / session.
|
|
subs_snapshot = [
|
|
dict(
|
|
wallet=s.wallet_address,
|
|
hl_api_key=s.hl_api_key,
|
|
leverage=s.leverage,
|
|
position_size_usd=s.position_size_usd,
|
|
take_profit_pct=s.take_profit_pct,
|
|
stop_loss_pct=s.stop_loss_pct,
|
|
min_confidence=s.min_confidence,
|
|
)
|
|
for s in subscribers
|
|
]
|
|
post_id = post.id
|
|
post_confidence = post.ai_confidence or 0
|
|
|
|
tasks = [
|
|
_execute_for_subscriber(sub, post_id, post_confidence, asset, side)
|
|
for sub in subs_snapshot
|
|
]
|
|
await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
|
|
async def _execute_for_subscriber(
|
|
sub: dict,
|
|
post_id: int,
|
|
post_confidence: int,
|
|
asset: str,
|
|
side: str,
|
|
) -> None:
|
|
wallet = sub["wallet"]
|
|
if not sub["hl_api_key"]:
|
|
logger.warning("Subscriber %s has no HL API key, skipping", wallet)
|
|
return
|
|
if post_confidence < sub["min_confidence"]:
|
|
logger.info("Sub %s filters out post %d: conf %d < user min %d",
|
|
wallet, post_id, post_confidence, sub["min_confidence"])
|
|
return
|
|
|
|
try:
|
|
api_key = decrypt_api_key(sub["hl_api_key"])
|
|
except Exception as exc:
|
|
logger.error("Cannot decrypt key for %s: %s", wallet, exc)
|
|
return
|
|
|
|
# Each subscriber gets its own session — AsyncSession is NOT concurrency-safe.
|
|
async with AsyncSessionLocal() as db:
|
|
try:
|
|
trader = HyperliquidTrader(
|
|
api_private_key=api_key,
|
|
account_address=wallet,
|
|
leverage=sub["leverage"],
|
|
mainnet=settings.hl_mainnet,
|
|
)
|
|
|
|
open_positions = await trader.get_open_positions()
|
|
if any(p.get('coin') == asset for p in open_positions):
|
|
logger.info("Subscriber %s already has open %s position, skipping", wallet, asset)
|
|
return
|
|
|
|
result = await trader.open_position(asset, side, sub["position_size_usd"])
|
|
entry_price = result.get('fill_price', 0.0)
|
|
|
|
trade = BotTrade(
|
|
asset=asset,
|
|
side=side,
|
|
entry_price=entry_price,
|
|
wallet_address=wallet,
|
|
trigger_post_id=post_id,
|
|
opened_at=datetime.now(timezone.utc).replace(tzinfo=None),
|
|
hl_order_id=result.get('order_id'),
|
|
)
|
|
db.add(trade)
|
|
await db.commit()
|
|
await db.refresh(trade)
|
|
|
|
logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)",
|
|
side, asset, wallet, entry_price, trade.id)
|
|
|
|
from app.services.tp_sl_monitor import register_trade
|
|
register_trade(
|
|
trade_id=trade.id,
|
|
wallet=wallet,
|
|
api_key=api_key,
|
|
leverage=sub["leverage"],
|
|
asset=asset,
|
|
side=side,
|
|
entry_price=entry_price,
|
|
take_profit_pct=sub["take_profit_pct"],
|
|
stop_loss_pct=sub["stop_loss_pct"],
|
|
)
|
|
|
|
asyncio.create_task(_close_after_hold(
|
|
trade.id, api_key, sub["leverage"], asset, wallet
|
|
))
|
|
|
|
except Exception as e:
|
|
logger.error("Trade execution failed for %s: %s", wallet, e)
|
|
|
|
|
|
async def close_and_finalize(
|
|
trade_id: int,
|
|
api_key: str,
|
|
leverage: int,
|
|
asset: str,
|
|
wallet: str,
|
|
reason: str = "max_hold",
|
|
) -> None:
|
|
"""
|
|
Close a live HL position and write exit_price/pnl to BotTrade.
|
|
Race-safe: a conditional UPDATE `closed_at = now WHERE closed_at IS NULL`
|
|
lets exactly one caller win; losers return silently. Also wraps in a
|
|
per-trade asyncio.Lock to prevent us even *attempting* the HL API call twice.
|
|
"""
|
|
async with _lock_for(trade_id):
|
|
async with AsyncSessionLocal() as db:
|
|
try:
|
|
now_naive = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
# Atomic claim: only one caller sets closed_at.
|
|
claim = await db.execute(
|
|
update(BotTrade)
|
|
.where(BotTrade.id == trade_id)
|
|
.where(BotTrade.closed_at.is_(None))
|
|
.values(closed_at=now_naive)
|
|
)
|
|
await db.commit()
|
|
if claim.rowcount == 0:
|
|
return # someone else closed it
|
|
|
|
# Reload the row we just claimed
|
|
result = await db.execute(select(BotTrade).where(BotTrade.id == trade_id))
|
|
trade = result.scalar_one()
|
|
|
|
sub_res = await db.execute(
|
|
select(Subscription).where(Subscription.wallet_address == wallet)
|
|
)
|
|
sub = sub_res.scalar_one_or_none()
|
|
size_usd = sub.position_size_usd if sub else 20.0
|
|
|
|
trader = HyperliquidTrader(
|
|
api_private_key=api_key,
|
|
account_address=wallet,
|
|
leverage=leverage,
|
|
mainnet=settings.hl_mainnet,
|
|
)
|
|
close_result = await trader.close_position(asset)
|
|
exit_price = close_result.get('fill_price', 0.0)
|
|
|
|
now_aware = datetime.now(timezone.utc)
|
|
opened_aware = trade.opened_at.replace(tzinfo=timezone.utc)
|
|
hold_secs = int((now_aware - opened_aware).total_seconds())
|
|
pct = ((exit_price - trade.entry_price) / trade.entry_price) if trade.entry_price else 0.0
|
|
signed_pct = pct if trade.side == 'long' else -pct
|
|
pnl_usd = size_usd * signed_pct * leverage
|
|
|
|
trade.exit_price = exit_price
|
|
trade.pnl_usd = round(pnl_usd, 2)
|
|
trade.hold_seconds = hold_secs
|
|
# closed_at already set by the atomic UPDATE
|
|
await db.commit()
|
|
|
|
# Clean up TP/SL + lock to avoid leaking memory
|
|
from app.services.tp_sl_monitor import unregister
|
|
unregister(trade_id)
|
|
_close_locks.pop(trade_id, None)
|
|
|
|
logger.info("Closed %s %s for %s @ %.2f PnL=%.2f (reason=%s)",
|
|
trade.side, asset, wallet, exit_price, pnl_usd, reason)
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to close trade %d: %s", trade_id, e)
|
|
|
|
|
|
async def _close_after_hold(
|
|
trade_id: int, api_key: str, leverage: int, asset: str, wallet: str
|
|
) -> None:
|
|
await asyncio.sleep(MAX_HOLD_SECONDS)
|
|
await close_and_finalize(trade_id, api_key, leverage, asset, wallet, reason="max_hold")
|