240 lines
8.5 KiB
Python
Executable File
240 lines
8.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""GARP trading decision engine — multiplayer aware."""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
|
|
import yfinance as yf
|
|
|
|
import game_engine
|
|
import scanner
|
|
|
|
MAX_POSITIONS = 15
|
|
MAX_POSITION_PCT = 0.10
|
|
RSI_BUY_LIMIT = 70
|
|
RSI_SELL_LIMIT = 80
|
|
NEAR_HIGH_PCT = 2.0
|
|
|
|
LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "logs")
|
|
|
|
|
|
def log_decision(action, ticker, reason, details=None):
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
entry = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"action": action,
|
|
"ticker": ticker,
|
|
"reason": reason,
|
|
"details": details or {},
|
|
}
|
|
log_file = os.path.join(LOG_DIR, f"{datetime.now().strftime('%Y-%m-%d')}.json")
|
|
logs = []
|
|
if os.path.exists(log_file):
|
|
with open(log_file) as f:
|
|
logs = json.load(f)
|
|
logs.append(entry)
|
|
with open(log_file, "w") as f:
|
|
json.dump(logs, f, indent=2, default=str)
|
|
return entry
|
|
|
|
|
|
def update_all_prices(game_id, username):
|
|
"""Update current prices for all held positions."""
|
|
p = game_engine.get_portfolio(game_id, username)
|
|
updated = []
|
|
for ticker in p["positions"]:
|
|
try:
|
|
stock = yf.Ticker(ticker)
|
|
hist = stock.history(period="5d")
|
|
if not hist.empty:
|
|
price = float(hist["Close"].iloc[-1])
|
|
game_engine.update_price(game_id, username, ticker, price)
|
|
updated.append((ticker, price))
|
|
except Exception as e:
|
|
print(f" Warning: Could not update {ticker}: {e}")
|
|
return updated
|
|
|
|
|
|
def check_sell_signals(game_id, username):
|
|
"""Check existing positions for sell signals."""
|
|
p = game_engine.get_portfolio(game_id, username)
|
|
sells = []
|
|
|
|
if not p["positions"]:
|
|
return sells
|
|
|
|
latest_scan = scanner.load_latest_scan()
|
|
scan_tickers = set()
|
|
if latest_scan:
|
|
scan_tickers = {c["ticker"] for c in latest_scan.get("candidates", [])}
|
|
|
|
for ticker, pos in list(p["positions"].items()):
|
|
sell_reason = None
|
|
|
|
if pos["current_price"] <= pos.get("trailing_stop", 0):
|
|
sell_reason = f"Trailing stop hit (stop={pos.get('trailing_stop', 0):.2f}, price={pos['current_price']:.2f})"
|
|
|
|
if not sell_reason:
|
|
try:
|
|
stock = yf.Ticker(ticker)
|
|
hist = stock.history(period="3mo")
|
|
if not hist.empty and len(hist) >= 15:
|
|
rsi = scanner.compute_rsi(hist["Close"].values)
|
|
if rsi and rsi > RSI_SELL_LIMIT:
|
|
sell_reason = f"RSI overbought ({rsi:.1f} > {RSI_SELL_LIMIT})"
|
|
except:
|
|
pass
|
|
|
|
if not sell_reason and latest_scan and ticker not in scan_tickers:
|
|
sell_reason = f"No longer passes GARP filter"
|
|
|
|
if sell_reason:
|
|
result = game_engine.sell(game_id, username, ticker, price=pos["current_price"], reason=sell_reason)
|
|
log_entry = log_decision("SELL", ticker, sell_reason, result)
|
|
sells.append(log_entry)
|
|
print(f" SELL {ticker}: {sell_reason}")
|
|
|
|
return sells
|
|
|
|
|
|
def check_buy_signals(game_id, username, candidates=None):
|
|
"""Check scan candidates for buy signals."""
|
|
p = game_engine.get_portfolio(game_id, username)
|
|
buys = []
|
|
|
|
if p["num_positions"] >= MAX_POSITIONS:
|
|
print(f" Max positions reached ({MAX_POSITIONS}), skipping buys")
|
|
return buys
|
|
|
|
if candidates is None:
|
|
latest_scan = scanner.load_latest_scan()
|
|
if not latest_scan:
|
|
print(" No scan data available")
|
|
return buys
|
|
candidates = latest_scan.get("candidates", [])
|
|
|
|
position_size = p["total_value"] / MAX_POSITIONS
|
|
max_per_position = p["total_value"] * MAX_POSITION_PCT
|
|
existing_tickers = set(p["positions"].keys())
|
|
|
|
for c in candidates:
|
|
if p["num_positions"] + len(buys) >= MAX_POSITIONS:
|
|
break
|
|
|
|
ticker = c["ticker"]
|
|
if ticker in existing_tickers:
|
|
continue
|
|
|
|
rsi = c.get("rsi")
|
|
if rsi and rsi > RSI_BUY_LIMIT:
|
|
log_decision("SKIP", ticker, f"RSI too high ({rsi:.1f} > {RSI_BUY_LIMIT})")
|
|
continue
|
|
|
|
pct_from_high = c.get("pct_from_52wk_high", 0)
|
|
if pct_from_high < NEAR_HIGH_PCT:
|
|
log_decision("SKIP", ticker, f"Too close to 52wk high ({pct_from_high:.1f}% away)")
|
|
continue
|
|
|
|
price = c["price"]
|
|
# Refresh cash from current portfolio state
|
|
current_p = game_engine.get_portfolio(game_id, username)
|
|
amount = min(position_size, max_per_position, current_p["cash"])
|
|
if amount < price:
|
|
continue
|
|
|
|
shares = int(amount / price)
|
|
if shares < 1:
|
|
continue
|
|
|
|
reason = (f"GARP signal: PE={c['trailing_pe']}, FwdPE={c['forward_pe']}, "
|
|
f"RevGr={c['revenue_growth']}%, EPSGr={c['earnings_growth']}%, RSI={rsi}")
|
|
|
|
result = game_engine.buy(game_id, username, ticker, shares, price, reason=reason)
|
|
if result["success"]:
|
|
log_entry = log_decision("BUY", ticker, reason, result)
|
|
buys.append(log_entry)
|
|
sector = c.get("sector", "Unknown")
|
|
print(f" BUY {ticker} ({sector}): {shares} shares @ ${price:.2f} = ${shares * price:,.2f}")
|
|
else:
|
|
# Log the specific reason for failure (sector cap, cash reserve, etc.)
|
|
error = result.get('error', 'unknown')
|
|
log_decision("SKIP", ticker, f"Buy failed: {error}")
|
|
if "sector" in error.lower() or "cash reserve" in error.lower():
|
|
print(f" SKIP {ticker}: {error}")
|
|
|
|
return buys
|
|
|
|
|
|
def check_rebalance(game_id, username):
|
|
"""Check if rebalancing is needed and optionally execute."""
|
|
rebalance_result = game_engine.rebalance_portfolio(game_id, username, dry_run=True)
|
|
|
|
if rebalance_result["violations"]:
|
|
print("\n Portfolio violations detected:")
|
|
for violation in rebalance_result["violations"]:
|
|
print(f" - {violation}")
|
|
|
|
if rebalance_result["actions"]:
|
|
print(f" Recommended rebalance actions:")
|
|
for action in rebalance_result["actions"]:
|
|
print(f" - {action['action']} {action['shares']} {action['ticker']} @ ${action['price']:.2f} ({action['reason']})")
|
|
|
|
# Auto-execute rebalancing for serious violations
|
|
total_excess_pct = (rebalance_result["total_excess"] / game_engine.get_portfolio(game_id, username)["total_value"]) * 100
|
|
if total_excess_pct > 5.0: # Auto-rebalance if excess > 5% of portfolio
|
|
print(f" Auto-executing rebalance (excess: {total_excess_pct:.1f}% of portfolio)...")
|
|
exec_result = game_engine.rebalance_portfolio(game_id, username, dry_run=False)
|
|
return exec_result
|
|
else:
|
|
print(" No rebalancing needed")
|
|
|
|
return rebalance_result
|
|
|
|
|
|
def run_trading_logic(game_id, username, candidates=None):
|
|
"""Run full trading cycle for a player."""
|
|
print(f"\n--- Trading Logic [{username}@{game_id}] ---")
|
|
|
|
print("\nUpdating prices...")
|
|
updated = update_all_prices(game_id, username)
|
|
for ticker, price in updated:
|
|
print(f" {ticker}: ${price:.2f}")
|
|
|
|
print("\nChecking portfolio balance...")
|
|
rebalance_result = check_rebalance(game_id, username)
|
|
|
|
print("\nChecking sell signals...")
|
|
sells = check_sell_signals(game_id, username)
|
|
if not sells:
|
|
print(" No sell signals")
|
|
|
|
print("\nChecking buy signals...")
|
|
buys = check_buy_signals(game_id, username, candidates)
|
|
if not buys:
|
|
print(" No buy signals")
|
|
|
|
# Show current portfolio summary
|
|
p = game_engine.get_portfolio(game_id, username)
|
|
if p:
|
|
sectors = game_engine.get_sector_allocation(game_id, username)
|
|
cash_pct = (p["cash"] / p["total_value"]) * 100
|
|
print(f"\nPortfolio Summary:")
|
|
print(f" Total Value: ${p['total_value']:,.2f}")
|
|
print(f" Cash: ${p['cash']:,.2f} ({cash_pct:.1f}%)")
|
|
print(f" Positions: {p['num_positions']}")
|
|
if sectors:
|
|
print(f" Sector Allocation:")
|
|
for sector, pct in sorted(sectors.items(), key=lambda x: x[1], reverse=True):
|
|
print(f" {sector}: {pct:.1f}%")
|
|
|
|
return {"sells": sells, "buys": buys, "price_updates": len(updated), "rebalance": rebalance_result}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
gid = game_engine.get_default_game_id()
|
|
if gid:
|
|
run_trading_logic(gid, "case")
|
|
else:
|
|
print("No default game found. Run game_engine.py first.")
|