-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathfinal_engine.py
More file actions
301 lines (239 loc) Β· 11.2 KB
/
final_engine.py
File metadata and controls
301 lines (239 loc) Β· 11.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
import os
import time
import requests
import json
import ollama
from dotenv import load_dotenv
from web3 import Web3
# ==============================================================================
# 1. SYSTEM CONFIGURATION & SETUP
# ==============================================================================
load_dotenv()
print("π€ INITIALIZING TRIDENT HEDGE FUND ENGINE...")
# A. CREDENTIALS
PRIVATE_KEY = os.getenv("PRIVATE_KEY")
BASESCAN_KEY = os.getenv("BASE_SCAN")
CRYPTOPANIC_KEY = os.getenv("CRYPTOPANIC_API_KEY")
print("β
CREDENTIALS LOADED.")
print(f" - BaseScan Key: {BASESCAN_KEY[:4]}...{BASESCAN_KEY[-4:]}")
print(f" - CryptoPanic Key: {CRYPTOPANIC_KEY[:4]}...{CRYPTOPANIC_KEY[-4:]}")
# B. BLOCKCHAIN CONNECTION (Base Sepolia for Trading)
RPC_URL = os.getenv("RPC_URL")
w3 = Web3(Web3.HTTPProvider(RPC_URL))
account = w3.eth.account.from_key(PRIVATE_KEY)
my_address = account.address
print(f"π WALLET CONNECTED: {my_address}")
# C. ASSETS
USDC_MAINNET = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" # Watching Real Whales
WETH_TESTNET = "0x4200000000000000000000000000000000000006" # Trading Fake ETH
USDC_TESTNET = "0x036CbD53842c5426634e7929541eC2318f3dCF7e" # Trading Fake USDC
ROUTER_ADDRESS = "0x94cC0AaC535CCDB3C01d6787D6413C739ae12bc4" # Uniswap Router
router_abi = [
{
"inputs": [
{
"components": [
{"internalType": "address", "name": "tokenIn", "type": "address"},
{"internalType": "address", "name": "tokenOut", "type": "address"},
{"internalType": "uint24", "name": "fee", "type": "uint24"},
{"internalType": "address", "name": "recipient", "type": "address"},
{"internalType": "uint256", "name": "amountIn", "type": "uint256"},
{"internalType": "uint256", "name": "amountOutMinimum", "type": "uint256"},
{"internalType": "uint160", "name": "sqrtPriceLimitX96", "type": "uint160"}
],
"internalType": "struct IV3SwapRouter.ExactInputSingleParams",
"name": "params",
"type": "tuple"
}
],
"name": "exactInputSingle",
"outputs": [{"internalType": "uint256", "name": "amountOut", "type": "uint256"}],
"stateMutability": "payable",
"type": "function"
}
]
print(f"π Loaded Router ABI. {len(router_abi)} items")
# Create the Contract Object
router_contract = w3.eth.contract(address=ROUTER_ADDRESS, abi=router_abi)
def get_ai_sentiment():
print("\n[1/3] π§ AI ANALYST (Fundamental Analysis)")
# Using the V2 Developer URL structure you confirmed earlier
url = f"https://cryptopanic.com/api/developer/v2/posts/?auth_token={CRYPTOPANIC_KEY}&kind=news&filter=rising&public=true"
try:
response = requests.get(url, headers={"User-Agent": "TridentBot/1.0"})
if response.status_code != 200:
print(f" β API HTTP ERROR: {response.status_code}")
print(f" β MSG: {response.text[:100]}...") # Print first 100 chars
return 0
data = response.json()
if 'results' not in data:
print(f" β οΈ UNEXPECTED DATA FORMAT: {data.keys()}")
return 0
headlines = [f"- {p['title']}" for p in data['results'][:5]]
if not headlines:
print(" β οΈ No rising news found.")
return 0
news_block = "\n".join(headlines)
print(f" π Reading {len(headlines)} headlines...")
# Analyze with Llama
prompt = f"""
Analyze these crypto headlines and output a JSON with a 'score' (-100 to 100).
Headlines: {news_block}
Output ONLY JSON: {{"score": 0, "reason": "..."}}
"""
ai_res = ollama.chat(model='llama3.2', messages=[{'role': 'user', 'content': prompt}])
# Clean AI Output (Find JSON)
content = ai_res['message']['content']
start, end = content.find('{'), content.rfind('}') + 1
result = json.loads(content[start:end])
score = int(result['score'])
print(f" β
AI VERDICT: {score}/100 ({result.get('reason', 'No reason')})")
return score
except Exception as e:
print(f" β AI PROCESSING FAILED: {e}")
return 0
# ==============================================================================
# 3. SIGNAL 2: WHALE WATCHER (On-Chain Data)
# ==============================================================================
def get_whale_score():
print("[2/3] π³ WHALE WATCHER (Omnichain V2 Analysis)")
BASE_CHAIN_ID = 8453
url = f"https://api.etherscan.io/v2/api?chainid={BASE_CHAIN_ID}&module=account&action=tokentx&contractaddress={USDC_MAINNET}&page=1&offset=20&sort=desc&apikey={BASESCAN_KEY}"
# RETRY LOGIC: Try 3 times before failing
max_retries = 3
for attempt in range(max_retries):
try:
res = requests.get(url).json()
# Check for Success
if res['status'] == '1':
# SUCCESS! Process the data
transfers = res.get('result', [])
whale_moves = 0
volume = 0
for tx in transfers:
amount = int(tx['value']) / 1_000_000
if amount > 100_000:
whale_moves += 1
volume += amount
print(f" found {whale_moves} Whales moving ${volume:,.0f} USDC.")
if whale_moves > 3: return -50
if whale_moves > 0: return 25
return 50
else:
# FAILED (NOTOK). Print why and wait.
print(f" β οΈ Attempt {attempt+1}/{max_retries} Failed: {res['message']}")
if attempt < max_retries - 1:
time.sleep(2) # Wait 2 seconds and try again
continue
except Exception as e:
print(f" β Connection Error: {e}")
time.sleep(2)
# If we get here, all 3 attempts failed
print(" β WHALE WATCH FAILED (Using Neutral Score)")
return 0
# ==============================================================================
# 4. SIGNAL 3: MARKET MOOD (Fear & Greed API)
# ==============================================================================
def get_market_mood():
print("[3/3] π MARKET MOOD (Technical Analysis)")
try:
res = requests.get("https://api.alternative.me/fng/")
data = res.json()['data'][0]
value = int(data['value'])
print(f" Global Sentiment: {value}/100 ({data['value_classification']})")
return (value - 50) * 2
except:
return 0
# ==============================================================================
# 5. EXECUTION ENGINE (BIDIRECTIONAL)
# ==============================================================================
def execute_swap(side):
print(f"\nπ€ PREPARING {side.upper()}: {'USDC -> ETH' if side == 'buy' else 'ETH -> USDC'}")
# 1. SETUP ASSETS
# We use WETH because Uniswap V3 doesn't trade raw ETH, it trades Wrapped ETH
token_in = USDC_TESTNET if side == "buy" else WETH_TESTNET
token_out = WETH_TESTNET if side == "buy" else USDC_TESTNET
# 2. AMOUNT LOGIC
# Buy: Swap 1 USDC (6 decimals)
# Sell: Swap 0.0001 ETH (18 decimals)
amount_in = 1000 if side == "buy" else w3.to_wei(0.0001, "ether")
try:
# --- BUY SIDE EXTRA STEP: APPROVAL ---
if side == "buy":
print(" π Checking USDC Allowance...")
# Minimal ABI for ERC20 Approval
erc20_abi = [{"constant":False,"inputs":[{"name":"_spender","type":"address"},{"name":"_value","type":"uint256"}],"name":"approve","outputs":[{"name":"","type":"bool"}],"type":"function"}]
usdc_contract = w3.eth.contract(address=USDC_TESTNET, abi=erc20_abi)
# 1. Build Approval Transaction
approve_txn = usdc_contract.functions.approve(ROUTER_ADDRESS, int(amount_in)).build_transaction({
"from": my_address,
"gas": 100000,
"maxFeePerGas": w3.to_wei("2", "gwei"),
"maxPriorityFeePerGas": w3.to_wei("1", "gwei"),
"nonce": w3.eth.get_transaction_count(my_address),
"chainId": 84532
})
# 2. Sign & Send Approval
print(" π Broadcasting Approval...")
signed_approve = w3.eth.account.sign_transaction(approve_txn, private_key=PRIVATE_KEY)
tx_hash_approve = w3.eth.send_raw_transaction(signed_approve.raw_transaction)
print(f" Wait for Approval: {w3.to_hex(tx_hash_approve)}...")
w3.eth.wait_for_transaction_receipt(tx_hash_approve)
print(" β
Approved! Now Swapping...")
# --- SWAP LOGIC (SAME FOR BOTH) ---
params = (
token_in,
token_out,
3000, # Fee Tier
my_address, # Recipient
int(amount_in), # Amount In
0, # Min Amount Out
0 # Price Limit
)
current_nonce = w3.eth.get_transaction_count(my_address)
swap_txn = router_contract.functions.exactInputSingle(params).build_transaction({
"from": my_address,
"value": 0 if side == "buy" else int(amount_in),
"gas": 300000,
"maxFeePerGas": w3.to_wei("2", "gwei"),
"maxPriorityFeePerGas": w3.to_wei("1", "gwei"),
"nonce": current_nonce,
"chainId": 84532
})
print(" βοΈ Signing Swap...")
signed_txn = w3.eth.account.sign_transaction(swap_txn, private_key=PRIVATE_KEY)
print(" π Broadcasting Swap...")
tx_hash = w3.eth.send_raw_transaction(signed_txn.raw_transaction)
print(f" β
Sent! Hash: {w3.to_hex(tx_hash)}")
# Wait for receipt
w3.eth.wait_for_transaction_receipt(tx_hash)
print(" π SWAP CONFIRMED.")
except Exception as e:
print(f"β EXECUTION ERROR: {e}")
def run_trident_strategy():
print("\n" + "="*50)
print("π± STARTING TRIDENT STRATEGY CYCLE")
print("="*50)
# 1. GATHER SIGNALS
s1_ai = get_ai_sentiment() # Weight: 40%
s2_whale = get_whale_score() # Weight: 30%
s3_mood = get_market_mood() # Weight: 30%
# 2. CALCULATE CONFIDENCE
final_score = (s1_ai * 0.4) + (s2_whale * 0.3) + (s3_mood * 0.3)
print("\n" + "-"*30)
print(f"π FINAL TRIDENT SCORE: {final_score:.2f} / 100")
print("-"*30)
# 3. DECISION MATRIX
if final_score < 100:
print("π DECISION: STRONG BUY")
execute_swap("buy")
elif final_score < -40:
print("jw DECISION: PANIC SELL")
execute_swap("sell")
else:
print("β DECISION: HOLD")
if __name__ == "__main__":
while True:
run_trident_strategy()
print("\nπ€ Sleeping for 60 seconds...")
time.sleep(60)