Claude AI Integration

import anthropic
import json
import requests
import os

CLAW_API_KEY = os.environ["CLAW_API_KEY"]
CLAW_BASE = "https://tickatlas.com/v1"

def call_tickatlas(endpoint: str, params: dict) -> dict:
    resp = requests.get(
        f"{CLAW_BASE}/{endpoint}",
        headers={"X-API-Key": CLAW_API_KEY},
        params=params,
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

# Define TickAtlas tools for Claude
tools = [
    {
        "name": "get_indicator",
        "description": "Get a real-time technical indicator value for a trading symbol.",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "e.g. EURUSD, XAUUSD, BTCUSD"},
                "indicator": {"type": "string", "description": "e.g. RSI_14, MACD_hist, SMA_200"},
                "timeframe": {"type": "string", "enum": ["M5","M15","M30","H1","H4","D1"]}
            },
            "required": ["symbol", "indicator", "timeframe"]
        }
    },
    {
        "name": "get_market_summary",
        "description": "Get AI-powered market analysis: bias, confidence, and narrative.",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string"},
                "timeframe": {"type": "string", "enum": ["M5","M15","M30","H1","H4","D1"]}
            },
            "required": ["symbol", "timeframe"]
        }
    },
    {
        "name": "get_all_indicators",
        "description": "Get all 42 indicators for a symbol at once.",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string"},
                "timeframe": {"type": "string", "enum": ["M5","M15","M30","H1","H4","D1"]}
            },
            "required": ["symbol", "timeframe"]
        }
    },
    {
        "name": "screen_market",
        "description": "Scan all available symbols to find those matching indicator criteria.",
        "input_schema": {
            "type": "object",
            "properties": {
                "indicator": {"type": "string"},
                "timeframe": {"type": "string"},
                "max": {"type": "number"},
                "min": {"type": "number"}
            },
            "required": ["indicator", "timeframe"]
        }
    }
]

client = anthropic.Anthropic()

def handle_tool_use(tool_name: str, tool_input: dict) -> str:
    """Route Claude's tool call to the correct TickAtlas endpoint."""
    if tool_name == "get_indicator":
        result = call_tickatlas("indicator", tool_input)
    elif tool_name == "get_market_summary":
        result = call_tickatlas("summary", tool_input)
    elif tool_name == "get_all_indicators":
        result = call_tickatlas("indicators", tool_input)
    elif tool_name == "screen_market":
        params = {k: v for k, v in tool_input.items() if v is not None}
        result = call_tickatlas("screener", params)
    else:
        result = {"error": f"Unknown tool: {tool_name}"}
    return json.dumps(result)

def chat(user_message: str):
    messages = [{"role": "user", "content": user_message}]

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system=(
            "You are a professional forex analyst. Use the TickAtlas tools "
            "to fetch real-time data before answering. Always cite specific "
            "indicator values and explain what they mean for the trader."
        ),
        tools=tools,
        messages=messages,
    )

    while response.stop_reason == "tool_use":
        messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = handle_tool_use(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

        messages.append({"role": "user", "content": tool_results})

        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

    for block in response.content:
        if hasattr(block, "text"):
            print(block.text)

chat("Is EURUSD bullish on H1? What about XAUUSD?")

pip install mcp requests

# mcp_tickatlas.py — MCP Server for TickAtlas
# Use with Claude Desktop or any MCP-compatible client

import json
import requests
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("tickatlas")
CLAW_BASE = "https://tickatlas.com/v1"
API_KEY = "YOUR_API_KEY"  # Set via environment variable in production

def _fetch(endpoint: str, params: dict) -> dict:
    resp = requests.get(
        f"{CLAW_BASE}/{endpoint}",
        headers={"X-API-Key": API_KEY},
        params=params,
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

@mcp.tool()
def indicator(symbol: str, indicator: str, timeframe: str = "H1") -> str:
    """Get a technical indicator value (RSI, MACD, SMA, etc.) for a symbol."""
    data = _fetch("indicator", {"symbol": symbol, "indicator": indicator, "timeframe": timeframe})
    return json.dumps(data, indent=2)

@mcp.tool()
def summary(symbol: str, timeframe: str = "H1") -> str:
    """Get AI market analysis with bias, confidence, and narrative."""
    data = _fetch("summary", {"symbol": symbol, "timeframe": timeframe})
    return json.dumps(data, indent=2)

@mcp.tool()
def screener(indicator: str, timeframe: str = "H1", max_val: float = None, min_val: float = None) -> str:
    """Scan all symbols by indicator criteria."""
    params = {"indicator": indicator, "timeframe": timeframe}
    if max_val is not None: params["max"] = max_val
    if min_val is not None: params["min"] = min_val
    data = _fetch("screener", params)
    return json.dumps(data, indent=2)

@mcp.tool()
def heatmap(timeframe: str = "H1") -> str:
    """Get a market heatmap showing all symbols with their current performance."""
    data = _fetch("heatmap", {"timeframe": timeframe})
    return json.dumps(data, indent=2)

@mcp.tool()
def calendar() -> str:
    """Get the economic calendar with upcoming high-impact events."""
    data = _fetch("calendar", {})
    return json.dumps(data, indent=2)

if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "tickatlas": {
      "command": "python",
      "args": ["mcp_tickatlas.py"],
      "env": {
        "CLAW_API_KEY": "your_api_key_here"
      }
    }
  }
}

# Autonomous trading analysis agent
# Claude analyzes the market on a schedule and generates reports

import anthropic
import json
import schedule
import time

client = anthropic.Anthropic()

def run_analysis():
    """Let Claude autonomously analyze the market and produce a report."""
    response = chat(
        "Perform a complete market analysis:\n"
        "1. Check RSI on H1 for EURUSD, GBPUSD, USDJPY, XAUUSD, BTCUSD\n"
        "2. Get the AI summary for any pair with RSI below 35 or above 65\n"
        "3. Run a screener for oversold (RSI < 30) instruments\n"
        "4. Generate a structured report with your findings and trade ideas\n\n"
        "Be specific about values and explain your reasoning."
    )
    with open(f"reports/market_{time.strftime('%Y%m%d_%H%M')}.md", "w") as f:
        f.write(response)

# Run every hour during market hours
schedule.every().hour.at(":05").do(run_analysis)

while True:
    schedule.run_pending()
    time.sleep(30)