louiscklaw/004_comission
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update,

Browse Source
This commit is contained in:
louiscklaw
2025-01-31 19:28:53 +08:00
parent 72bacdd6b5
commit 843c590c8b
19 changed files with 1522 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
calvin312321/meta.md Normal file
View File

@@ -0,0 +1,7 @@
---
tags: [rude]
---
# calvin312321
02-jan-2025 rude, clever (not in a good way) and a lot of questions

160
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/.gitignore vendored Normal file
View File

@@ -0,0 +1,160 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
.idea/

26
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/.run/webhook_app.run.xml Normal file
View File

@@ -0,0 +1,26 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="webhook_app"
type="PythonConfigurationType" factoryName="Python">
<module name="TWS-orders-placement-via-Tradinview-webhooks"/>
<option name="INTERPRETER_OPTIONS" value=""/>
<option name="PARENT_ENVS" value="true"/>
<envs>
<env name="PYTHONUNBUFFERED" value="1"/>
</envs>
<option name="SDK_HOME" value=""/>
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$"/>
<option name="IS_MODULE_SDK" value="true"/>
<option name="ADD_CONTENT_ROOTS" value="true"/>
<option name="ADD_SOURCE_ROOTS" value="true"/>
<EXTENSION ID="PythonCoverageRunConfigurationExtension"
runner="coverage.py"/>
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/main.py"/>
<option name="PARAMETERS" value=""/>
<option name="SHOW_COMMAND_LINE" value="false"/>
<option name="EMULATE_TERMINAL" value="false"/>
<option name="MODULE_MODE" value="false"/>
<option name="REDIRECT_INPUT" value="false"/>
<option name="INPUT_FILE" value=""/>
<method v="2"/>
</configuration>
</component>

65
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/README.md Normal file
View File

@@ -0,0 +1,65 @@
## Placing orders to IB TWS via Tradingview alerts webhooks
Connect TradingView with Interactive Brokers to process automated signals
as entries and exits in an IB brokerage account.
#### Python version 3.10
### Configuring the alert Webhook and installing ngrock
You'll need to install `ngrock` (URL to the download
page- https://ngrok.com/download)
and at least Pro TradingView subscription for placing webhooks in alerts,
and redirect them to your localhost.
Please, do not forget to add Authtoken from ngrock
- https://dashboard.ngrok.com/get-started/your-authtoken
After that you'll be able to run ngrock server:
```shell
$ ngrok http 5000
```
Copy the URL from `Forwarding` line and paste it into the alert Webhook line.
Add `/webhook` to the following URL. <b>Note: that webhooks are now available from Premium plan on TradingView<b>
Then, add the message to the `Message` field in the Alert navigation and click
Save:
```json
{
"message": "YourMessage",
"symbol": "{{ticker}}",
"price": "{{close}}",
"timeframe": "{{interval}}"
}
```
---
### Requirements
To run the application, please do not forget to install the following
requirements. You can do it in your terminal via the following command:
```shell
$ pip3 install --requirement requirements.txt
```
---
### Run app
To run the app via terminal do not forget to change the directory
to `src`.
After that you can simply type this command in your terminal:
```shell
$ python3 app.py
# or
$ chmod +x app.py
$ ./app.py
```

5
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
# Core
Flask==2.1.2
ib_insync==0.9.70
sanic==22.3.2
websockets==10.0

8
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/src/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
"""
The flask application package.
"""
from flask import Flask
import src.app
app = Flask(__name__)

374
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/src/app_ib_insync.py Normal file
View File

@@ -0,0 +1,374 @@
#!/usr/bin/env python3
"""
Routes and views for the application.
"""
import os
import sys
import re
import asyncio
from ib_insync import IB, MarketOrder, ExecutionFilter
from sanic import Sanic, response
from sanic.exceptions import InvalidUsage
import logging
from contract import get_next_contract_month, contract_type_check
import time
# Create the Sanic app
app = Sanic("unique_app_name")
# Initialize the logger
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# IB constants
TWS_HOST = "127.0.0.1"
TWS_LIVE_PORT = 7496
TWS_PAPER_PORT = 7497
IBG_HOST = "127.0.0.1"
IBG_LIVE_PORT = 4001
IBG_PAPER_PORT = 4002
# Configuration: Set to True for paper trading, False for live trading
USE_PAPER_TRADING = True
# Configuration: Set to True for IB Gateway, False for TWS
USE_IB_GATEWAY = True
# Determine the correct host and port based on the configuration
if USE_IB_GATEWAY:
HOST = IBG_HOST
PORT = IBG_PAPER_PORT if USE_PAPER_TRADING else IBG_LIVE_PORT
else:
HOST = TWS_HOST
PORT = TWS_PAPER_PORT if USE_PAPER_TRADING else TWS_LIVE_PORT
# Initialize the IB connection
app_ib = IB()
# Check every minute if we need to reconnect to IB
async def check_if_reconnect() -> None:
global app_ib
logger.info("Checking if we need to reconnect...")
# Reconnect if needed
if app_ib is None or not app_ib.isConnected():
try:
logger.info("Reconnecting...")
if app_ib is not None:
app_ib.disconnect()
await app_ib.connectAsync(HOST, PORT, clientId=0) # Using clientId=0 as Master Client
app_ib.errorEvent += check_on_ib_error
logger.info("Successfully reconnected to TWS")
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
file_name = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
logger.error(f"Exception: {exc_type}, File: {file_name}, Line: {exc_tb.tb_lineno}")
logger.warning(f"Make sure TWS or Gateway is open with the correct port: {str(e)}")
# Helper function to wait for the order to be filled
async def wait_for_order_fill(trade, timeout=60):
start_time = time.time()
while trade.orderStatus.status not in ('Filled', 'Cancelled'):
if time.time() - start_time > timeout:
raise TimeoutError("Order fill timeout")
await app_ib.sleep(1)
return trade.orderStatus.status
# Request account updates
def request_account_updates(subscribe=True, accountCode=""):
app_ib.reqAccountUpdates(subscribe, accountCode)
# Event handler for order status
def on_order_status(trade):
logger.info(f"Order Status: OrderId={trade.order.orderId}, Status={trade.orderStatus.status}, Filled={trade.orderStatus.filled}, Remaining={trade.orderStatus.remaining}, AvgFillPrice={trade.orderStatus.avgFillPrice}")
# Event handler for account updates
def on_account_update(account, key, value, currency):
logger.info(f"Account update: Account={account}, Key={key}, Value={value}, Currency={currency}")
# Event handler for portfolio updates
def on_portfolio_update(account, contract, position, marketPrice, marketValue, averageCost, unrealizedPNL, realizedPNL):
logger.info(f"Portfolio update: Account={account}, Contract={contract}, Position={position}, MarketPrice={marketPrice}, MarketValue={marketValue}, AverageCost={averageCost}, UnrealizedPNL={unrealizedPNL}, RealizedPNL={realizedPNL}")
# Event handler for open orders
def on_open_order(trade):
logger.info(f"Open order: OrderId={trade.order.orderId}, Contract={trade.contract}, Order={trade.order}, OrderState={trade.orderState}")
# Event handler for IB error
def check_on_ib_error(reqId, error_code, error_string, contract):
logger.error(f"Error code: {error_code}, Message: {error_string}")
# Set event handlers
def set_event_handlers():
app_ib.orderStatusEvent += on_order_status
app_ib.accountSummaryEvent += on_account_update
app_ib.updatePortfolioEvent += on_portfolio_update
app_ib.openOrderEvent += on_open_order
app_ib.errorEvent += check_on_ib_error
# Periodic update function
async def periodic_update():
while True:
request_account_updates()
await asyncio.sleep(60) # Update every 60 seconds
@app.route('/portfolio', methods=['GET'])
async def portfolio(request) -> response.HTTPResponse:
if request.method == 'GET':
try:
await check_if_reconnect()
positions = await app_ib.reqPositions() # Use reqPositions to get real-time positions
if not positions:
logger.error("No positions retrieved from IB TWS API.")
return response.json({"status": "error", "message": "Failed to retrieve positions from IB TWS API."}, status=400)
portfolio_data = []
for position in positions:
contract = position.contract
avg_price = position.avgCost
quantity = position.position
unrealized_pnl = position.unrealizedPNL
realized_pnl = position.realizedPNL
portfolio_data.append({
"Instrument": contract.symbol,
"Average Price": avg_price,
"Positions": quantity,
"Unrealized P&L": unrealized_pnl,
"Realized P&L": realized_pnl
})
return response.json(portfolio_data)
except Exception as e:
logger.error(f"Error retrieving portfolio: {e}")
return response.json({"status": "error", "message": "Failed to retrieve portfolio"}, status=500)
return response.json({"status": "error", "message": "Invalid request method"}, status=405)
@app.route('/webhook', methods=['POST'])
async def webhook(request) -> response.HTTPResponse:
if request.method == 'POST':
try:
# Check if we need to reconnect with IB
await check_if_reconnect()
# Log the raw request body for debugging
raw_body = request.body.decode('utf-8')
logger.info(f"Received raw body: {raw_body}")
# Extract the required fields using regular expressions
action_pattern = r"order (?P<action>\w+) (?P<contracts>\d+) @ [\d\.]+ filled on (?P<ticker>\w+)"
position_pattern = r"New strategy position is (?P<position_size>-?\d+)"
action_match = re.search(action_pattern, raw_body)
position_match = re.search(position_pattern, raw_body)
if not action_match or not position_match:
logger.error("Failed to parse webhook message.")
return response.json({"status": "error", "message": "Failed to parse webhook message."}, status=400)
action = action_match.group("action").upper()
contracts = int(action_match.group("contracts"))
ticker = action_match.group("ticker")
position_size = int(position_match.group("position_size"))
logger.info(f"Parsed values - Action: {action}, Contracts: {contracts}, Ticker: {ticker}, Position Size: {position_size}")
# Determine the IB ticker based on the TradingView ticker
ib_ticker = None
if ticker == "SPX":
ib_ticker = "MES"
elif ticker == "NDX":
ib_ticker = "MNQ"
elif ticker == "HSI":
ib_ticker = "MHI"
elif ticker == "TOPIX":
ib_ticker = "MNTPX"
else:
logger.error(f"Invalid ticker: {ticker}")
return response.json({"status": "error", "message": "Invalid ticker"}, status=400)
# Map TradingView actions to IB actions
if action not in ["BUY", "SELL"]:
logger.error(f"Invalid action: {action}")
return response.json({"status": "error", "message": "Invalid action"}, status=400)
# Get the current contract month
current_contract_month = get_next_contract_month(ib_ticker)
current_contract = contract_type_check(ib_ticker, contract_month=current_contract_month)
if not current_contract:
return response.json({"status": "error", "message": "Invalid contract"}, status=400)
# 1. Place Initial Order: Execute the webhook order with contract month as of today
initial_order = MarketOrder(action, contracts, account=app_ib.wrapper.accounts[0])
logger.info(f"Placing initial order: {action} {contracts} contracts of {ib_ticker}")
# Place the order
try:
trade = app_ib.placeOrder(current_contract, initial_order)
logger.info(f"Order placed: {trade}")
# Log order status after placing
await asyncio.sleep(2) # Wait for a moment to allow the order to process
logger.info(f"Order status after placing: {trade.orderStatus.status}")
# Check for errors in the IB API
if trade.orderStatus.status == 'PendingSubmit':
logger.error("Order stuck in PendingSubmit.")
return response.json({"status": "error", "message": "Order stuck in PendingSubmit."}, status=400)
# 2. Wait for the initial order to be filled or cancelled
order_status = await wait_for_order_fill(trade)
logger.info(f"Order status after waiting: {order_status}")
if order_status != 'Filled':
logger.error("Order was not filled.")
return response.json({"status": "error", "message": "Order was not filled."}, status=400)
# 3. Retrieve Execution Details: Retrieve execution details for the order
exec_filter = ExecutionFilter(clientId=0)
executions = await app_ib.reqExecutions(exec_filter)
for execution in executions:
if execution.orderId == trade.order.orderId:
logger.info(f"Execution details: {execution}")
# 4. Retrieve Latest Position Details: Retrieve the current positions from IB TWS API
positions = await app_ib.reqPositions() # Use reqPositions to get real-time positions
if not positions:
logger.error("No positions retrieved from IB TWS API.")
return response.json({"status": "error", "message": "Failed to retrieve positions from IB TWS API."}, status=400)
current_position = sum(pos.position for pos in positions if pos.contract.symbol == ib_ticker)
ticker_positions = {pos.contract.symbol: pos.position for pos in positions}
logger.info(f"Current Positions: {ticker_positions}")
return response.json({"status": "success", "message": "Order placed and position details retrieved."})
except Exception as e:
logger.error(f"Error processing webhook: {e}")
return response.json({"status": "error", "message": f"Failed to process webhook: {str(e)}"}, status=500)
except InvalidUsage as e:
logger.error(f"Invalid JSON received: {e}")
return response.json({"status": "error", "message": "Invalid JSON format"}, status=400)
except Exception as e:
logger.error(f"Error processing webhook: {e}")
return response.json({"status": "error", "message": f"Failed to process webhook: {str(e)}"}, status=500)
return response.json({"status": "error", "message": "Invalid request method"}, status=405)
@app.route('/adjust_positions', methods=['POST'])
async def adjust_positions(request) -> response.HTTPResponse:
if request.method == 'POST':
try:
# Check if we need to reconnect with IB
await check_if_reconnect()
# Log the raw request body for debugging
raw_body = request.body.decode('utf-8')
logger.info(f"Received raw body: {raw_body}")
# Extract the required fields using regular expressions
action_pattern = r"order (?P<action>\w+) (?P<contracts>\d+) @ [\d\.]+ filled on (?P<ticker>\w+)"
position_pattern = r"New strategy position is (?P<position_size>-?\d+)"
action_match = re.search(action_pattern, raw_body)
position_match = re.search(position_pattern, raw_body)
if not action_match or not position_match:
logger.error("Failed to parse webhook message.")
return response.json({"status": "error", "message": "Failed to parse webhook message."}, status=400)
action = action_match.group("action").upper()
contracts = int(action_match.group("contracts"))
ticker = action_match.group("ticker")
position_size = int(position_match.group("position_size"))
logger.info(f"Parsed values - Action: {action}, Contracts: {contracts}, Ticker: {ticker}, Position Size: {position_size}")
# Determine the IB ticker based on the TradingView ticker
ib_ticker = None
if ticker == "SPX":
ib_ticker = "MES"
elif ticker == "NDX":
ib_ticker = "MNQ"
elif ticker == "HSI":
ib_ticker = "MHI"
elif ticker == "TOPIX":
ib_ticker = "MNTPX"
else:
logger.error(f"Invalid ticker: {ticker}")
return response.json({"status": "error", "message": "Invalid ticker"}, status=400)
# Get the current contract month
current_contract_month = get_next_contract_month(ib_ticker)
correct_contract = contract_type_check(ib_ticker, contract_month=current_contract_month)
rollover_needed = get_next_contract_month(ib_ticker, check_rollover=True)
positions = await app_ib.reqPositions()
positions_in_correct_month = [pos for pos in positions if pos.contract.lastTradeDateOrContractMonth == current_contract_month]
positions_not_in_correct_month = [pos for pos in positions if pos.contract.lastTradeDateOrContractMonth != current_contract_month]
# Contract month rollover if needed
if positions_not_in_correct_month:
logger.info(f"Contract month rollover needed: Adjusting positions to the correct contract month {current_contract_month}")
for pos in positions_not_in_correct_month:
close_action = "SELL" if pos.position > 0 else "BUY"
close_order = MarketOrder(close_action, abs(pos.position), account=app_ib.wrapper.accounts[0])
app_ib.placeOrder(pos.contract, close_order)
logger.info(f"Closed positions for outdated contract month {pos.contract.lastTradeDateOrContractMonth}")
# Open positions only for the difference between webhook and IB TWS positions
open_order_quantity = position_size - sum(pos.position for pos in positions_in_correct_month)
if open_order_quantity != 0:
open_order = MarketOrder("SELL" if open_order_quantity < 0 else "BUY", abs(open_order_quantity), account=app_ib.wrapper.accounts[0])
app_ib.placeOrder(correct_contract, open_order)
logger.info(f"Opened positions for the difference: {open_order_quantity} contracts of {ib_ticker} with contract month {current_contract_month}")
else:
logger.info(f"Contract month rollover check completed: Positions in IB TWS are using the correct contract month {current_contract_month}")
# Adjust Positions if Needed: Adjust positions to match the webhook
while True:
positions = await app_ib.reqPositions() # Use reqPositions to get real-time positions
if not positions:
logger.error("No positions retrieved from IB TWS API.")
return response.json({"status": "error", "message": "Failed to retrieve positions from IB TWS API."}, status=400)
current_position = sum(pos.position for pos in positions if pos.contract.symbol == ib_ticker)
ticker_positions = {pos.contract.symbol: pos.position for pos in positions}
logger.info(f"Current Positions: {ticker_positions}")
if current_position != position_size:
adjustment_quantity = position_size - current_position
adjustment_action = "SELL" if adjustment_quantity < 0 else "BUY"
adjustment_order = MarketOrder(adjustment_action, abs(adjustment_quantity), account=app_ib.wrapper.accounts[0])
logger.info(f"Adjusting Position: {adjustment_action} {abs(adjustment_quantity)} contracts of {ib_ticker}")
app_ib.placeOrder(correct_contract, adjustment_order)
logger.info(f"Position adjusted for {ib_ticker}: Before adjustment: {current_position}, After adjustment: {position_size}")
else:
logger.info(f"Position check completed: Positions in IB TWS match the positions from the webhook for contract month {current_contract_month}")
break
return response.json({"status": "success", "message": "Order placed and position adjusted if needed"})
except InvalidUsage as e:
logger.error(f"Invalid JSON received: {e}")
return response.json({"status": "error", "message": "Invalid JSON format"}, status=400)
except Exception as e:
logger.error(f"Error adjusting positions: {e}")
return response.json({"status": "error", "message": f"Failed to adjust positions: {str(e)}"}, status=500)
return response.json({"status": "error", "message": "Invalid request method"}, status=405)
if __name__ == '__main__':
try:
# Connect to IB on init
logger.info("Connecting to IB...")
app_ib.connect(HOST, PORT, clientId=0) # Using clientId=0 as Master Client
logger.info("Successfully Connected to IB")
# Set event handlers
set_event_handlers()
# Start the periodic update loop
asyncio.ensure_future(periodic_update())
app.run(port=8080)
except Exception as e:
logger.error(f"Failed to start the application: {e}")

105
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/src/contract.py Normal file
View File

@@ -0,0 +1,105 @@
from datetime import datetime, timedelta
from ib_insync import Contract
from logger import LOGGER as log
def get_next_contract_month(ticker: str, offset=0, check_rollover=False) -> str:
"""Returns the next contract month in 'YYYYMM' format based on the ticker and specified conditions."""
now = datetime.now()
def next_quarter_month(date, offset=0):
"""Get the next quarter month (March, June, September, December) with an optional offset."""
quarter_months = [3, 6, 9, 12]
month = date.month
year = date.year
# Find the next quarter month
next_month = min((m for m in quarter_months if m > month), default=quarter_months[0])
if next_month <= month:
year += 1
next_month_index = (quarter_months.index(next_month) + offset) % len(quarter_months)
next_month = quarter_months[next_month_index]
return year, next_month
def get_monday_before_second_friday(year, month):
"""Get the Monday before the second Friday of the given month and year."""
first_day = datetime(year, month, 1)
first_friday = first_day + timedelta(days=(4 - first_day.weekday() + 7) % 7) # First Friday
second_friday = first_friday + timedelta(days=7)
return second_friday - timedelta(days=second_friday.weekday() + 7)
def get_monday_before_third_friday(year, month):
"""Get the Monday before the third Friday of the given month and year."""
first_day = datetime(year, month, 1)
first_friday = first_day + timedelta(days=(4 - first_day.weekday() + 7) % 7) # First Friday
third_friday = first_friday + timedelta(days=14)
return third_friday - timedelta(days=third_friday.weekday() + 7)
def get_last_monday_before_second_last_trading_day(year, month):
"""Get the last Monday before the second last trading day of the given month and year."""
last_day = datetime(year, month + 1, 1) - timedelta(days=1)
second_last_trading_day = last_day
while second_last_trading_day.weekday() in (5, 6): # Skip weekends
second_last_trading_day -= timedelta(days=1)
second_last_trading_day -= timedelta(days=1)
while second_last_trading_day.weekday() in (5, 6): # Skip weekends
second_last_trading_day -= timedelta(days=1)
return second_last_trading_day - timedelta(days=second_last_trading_day.weekday() + 1)
if ticker in ["ES", "MES", "NQ", "MNQ"]:
year, month = next_quarter_month(now)
monday_before_third_friday = get_monday_before_third_friday(year, month)
if now >= monday_before_third_friday:
year, month = next_quarter_month(datetime(year, month, 1), 1)
if check_rollover:
return True # Indicate rollover is needed
return f"{year}{month:02d}"
elif ticker in ["TOPX", "MNTPX"]:
year, month = next_quarter_month(now)
monday_before_second_friday = get_monday_before_second_friday(year, month)
if now >= monday_before_second_friday:
year, month = next_quarter_month(datetime(year, month, 1), 1)
if check_rollover:
return True # Indicate rollover is needed
return f"{year}{month:02d}"
elif ticker in ["HSI", "MHI"]:
year, month = next_quarter_month(now)
last_monday = get_last_monday_before_second_last_trading_day(year, month)
if now >= last_monday:
year, month = next_quarter_month(datetime(year, month, 1), 1)
if check_rollover:
return True # Indicate rollover is needed
return f"{year}{month:02d}"
else:
log.error(f"Invalid ticker: {ticker}. Please check the alert message.")
return None
def contract_type_check(ticker: str, contract_month=None) -> Contract:
contract = Contract()
contract.symbol = ticker
if not contract_month:
contract_month = get_next_contract_month(ticker)
if ticker == "SPY":
contract.secType = "STK"
contract.currency = "USD"
contract.exchange = "ARCA"
elif ticker.startswith("ETH"):
contract.secType = "CRYPTO"
contract.currency = "USD"
contract.exchange = "PAXOS"
elif ticker.startswith("EUR"):
contract.secType = "CASH"
contract.currency = "USD"
contract.exchange = "IDEALPRO"
elif ticker in ["ES", "MES", "NQ", "MNQ", "HSI", "MHI", "TOPX", "MNTPX"]:
contract.secType = "FUT"
contract.currency = "USD" if ticker in ["ES", "MES", "NQ", "MNQ"] else "HKD" if ticker in ["HSI", "MHI"] else "JPY"
contract.exchange = "CME" if ticker in ["ES", "MES", "NQ", "MNQ"] else "HKFE" if ticker in ["HSI", "MHI"] else "OSE"
contract.lastTradeDateOrContractMonth = contract_month
else:
log.error(f"Invalid ticker: {ticker}. Please check the alert message.")
return None
return contract

24
calvin312321/quotation1/from_customer/TWS-orders-placement-via-Tradinview-webhooks/src/logger.py Normal file
View File

@@ -0,0 +1,24 @@
import logging
# Create and configure logger
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.INFO)
# Create console handler and set level to INFO
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
# Create file handler and set level to INFO
fh = logging.FileHandler('app.log')
fh.setLevel(logging.INFO)
# Create formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# Add formatter to handlers
ch.setFormatter(formatter)
fh.setFormatter(formatter)
# Add handlers to logger
LOGGER.addHandler(ch)
LOGGER.addHandler(fh)

41
calvin312321/quotation1/notes.md Normal file
View File

@@ -0,0 +1,41 @@
---
tags: trading view, IB
---
# NOTES
```markdown
hello, 想請問搵你代寫 python 點收費？唔該
要求功能：
由 TradingView Webhook 轉成 Interative Broker TWS API 落 order 然後收 order confirmation, 跟住 keep 住 request and receive portfolio positions 去做 verifications (FYI 目前已有 script, 成功做到 convert from webhook to order placing, 但唔知點解做唔到 interact with IB TWS API)
因為呢度 send 唔到 link, 請參考 GitHub
1. TWS-orders-placement-via-Tradinview-webhooks
2. ib_insync
```
> 用 TWS-orders-placement-via-Tradinview-webhooks 嘅 template，
> 然後再用 copilot 嚟幫我改做我嘅要求，已成功 place order，
> 但未能收到 IB TWS API 嘅 order confirmation 嚟俾個 script run 埋下半 part，
> 即係幫我 verify IB current positions 係咪等於我 webhook 個 new positions
> 我可以俾晒手頭上有嘅嘢俾你，你睇完先報價都冇問題，
> 但係我唔肯定係關 script 事定我自己未 install/setup 好所需要嘅嘢
hello, 想請問搵你代寫 python 點收費？唔該
要求功能：
1. 由 TradingView Webhook 轉成 Interative Broker TWS API 落 order
2. 然後收 order confirmation
3. 跟住 keep 住 request and receive portfolio positions 去做 verifications
4. FYI 目前已有 script, 成功做到 convert from webhook to order placing, 但唔知點解做唔到 interact with IB TWS API
因為呢度 send 唔到 link, 請參考 GitHub
1. TWS-orders-placement-via-Tradinview-webhooks
1. [https://github.com/EconLQ/TWS-orders-placement-via-Tradinview-webhooks](https://github.com/EconLQ/TWS-orders-placement-via-Tradinview-webhooks)
2. ib_insync(archived)
1. [https://github.com/erdewit/ib_insync](https://github.com/erdewit/ib_insync)