k
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
"""
|
|
Simulate ~30 days of bot trading history for a given wallet, using real historical
|
|
posts + the measured price_impact_{m5,m15,m1h} fields. Creates BotTrade rows as if
|
|
the bot had been running with default settings (size=$20, lev=3x, tp=2%, sl=1.5%).
|
|
|
|
Usage:
|
|
python simulate_trades.py <wallet_address>
|
|
|
|
Safe to re-run: skips posts already linked to a trade for the given wallet.
|
|
"""
|
|
import asyncio
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional, Tuple
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.database import AsyncSessionLocal
|
|
from app.models import BotTrade, Post
|
|
|
|
|
|
SIZE_USD = 20.0
|
|
LEVERAGE = 3
|
|
TP_PCT = 2.0 # close at +2% price move in position direction
|
|
SL_PCT = 1.5 # close at -1.5%
|
|
MIN_CONFIDENCE = 80
|
|
DAYS_BACK = 30
|
|
|
|
|
|
def simulate_exit(side: str, m5: Optional[float], m15: Optional[float], m1h: Optional[float]) -> Tuple[float, int, str]:
|
|
"""
|
|
Walk the three sampled points; exit at the first TP/SL crossing, else at m1h close.
|
|
Returns (signed_pct, hold_seconds, reason).
|
|
signed_pct = price move in position's favour (positive means profit, in raw %).
|
|
"""
|
|
samples = [(300, m5), (900, m15), (3600, m1h)]
|
|
for secs, pct in samples:
|
|
if pct is None:
|
|
continue
|
|
signed = pct if side == 'long' else -pct
|
|
if signed >= TP_PCT:
|
|
return TP_PCT, secs, 'take_profit'
|
|
if signed <= -SL_PCT:
|
|
return -SL_PCT, secs, 'stop_loss'
|
|
# No TP/SL hit — exit at whichever latest sample we have
|
|
final = m1h if m1h is not None else (m15 if m15 is not None else (m5 or 0.0))
|
|
signed = final if side == 'long' else -final
|
|
hold = 3600 if m1h is not None else (900 if m15 is not None else 300)
|
|
return signed, hold, 'max_hold'
|
|
|
|
|
|
async def main(wallet: str) -> None:
|
|
wallet = wallet.lower()
|
|
cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=DAYS_BACK)
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
# Get candidate posts
|
|
result = await db.execute(
|
|
select(Post)
|
|
.where(Post.published_at >= cutoff)
|
|
.where(Post.relevant == True) # noqa: E712
|
|
.where(Post.ai_confidence >= MIN_CONFIDENCE)
|
|
.where(Post.signal.in_(('buy', 'short')))
|
|
.where(Post.price_at_post.is_not(None))
|
|
.order_by(Post.published_at.asc())
|
|
)
|
|
posts = result.scalars().all()
|
|
|
|
# Skip posts already simulated for this wallet
|
|
existing = await db.execute(
|
|
select(BotTrade.trigger_post_id).where(BotTrade.wallet_address == wallet)
|
|
)
|
|
seen = {pid for (pid,) in existing.all() if pid is not None}
|
|
|
|
created = 0
|
|
total_pnl = 0.0
|
|
wins = 0
|
|
|
|
for p in posts:
|
|
if p.id in seen:
|
|
continue
|
|
asset = p.price_impact_asset or 'BTC'
|
|
side = 'long' if p.signal == 'buy' else 'short'
|
|
entry = p.price_at_post
|
|
if not entry:
|
|
continue
|
|
|
|
signed_pct, hold_secs, reason = simulate_exit(
|
|
side, p.price_impact_m5, p.price_impact_m15, p.price_impact_m1h
|
|
)
|
|
# exit_price reconstructed from signed_pct relative to side direction
|
|
raw_pct = signed_pct if side == 'long' else -signed_pct
|
|
exit_price = entry * (1 + raw_pct / 100.0)
|
|
pnl_usd = round(SIZE_USD * (signed_pct / 100.0) * LEVERAGE, 2)
|
|
|
|
opened = p.published_at
|
|
closed = opened + timedelta(seconds=hold_secs)
|
|
|
|
trade = BotTrade(
|
|
asset=asset,
|
|
side=side,
|
|
entry_price=round(entry, 2),
|
|
exit_price=round(exit_price, 2),
|
|
pnl_usd=pnl_usd,
|
|
hold_seconds=hold_secs,
|
|
trigger_post_id=p.id,
|
|
wallet_address=wallet,
|
|
opened_at=opened,
|
|
closed_at=closed,
|
|
hl_order_id=f'sim-{p.id}',
|
|
)
|
|
db.add(trade)
|
|
created += 1
|
|
total_pnl += pnl_usd
|
|
if pnl_usd > 0:
|
|
wins += 1
|
|
|
|
await db.commit()
|
|
|
|
wr = (wins / created * 100) if created else 0
|
|
print(f"✓ Simulated {created} trades for {wallet}")
|
|
print(f" Total PnL: ${total_pnl:+.2f}")
|
|
print(f" Win rate: {wr:.1f}% ({wins}/{created})")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if len(sys.argv) != 2:
|
|
print("Usage: python simulate_trades.py <wallet_address>")
|
|
sys.exit(1)
|
|
asyncio.run(main(sys.argv[1]))
|