Skip to main content

​
Overview

TWAP (Time-Weighted Average Price) orders enable you to split a large trade into smaller parts that execute at regular intervals. This strategy helps:
  • Reduce Price Impact — Avoid moving the market with large orders
  • Minimize Slippage — Spread execution risk over time
  • Dollar-Cost Averaging — Average out price fluctuations
  • Stealth Trading — Execute large positions without revealing full intent
TWAP orders are implemented as composable programmatic orders on the ComposableCoW framework.

​
Creating a TWAP Order

from cowdao_cowpy.composable.order_types.twap import Twap, TwapData, StartType, DurationType
from cowdao_cowpy.common.chains import Chain
from web3 import Web3

twap_data = TwapData(
    sell_token=Web3.to_checksum_address("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
    buy_token=Web3.to_checksum_address("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
    receiver=Web3.to_checksum_address("0xYourAddress"),
    sell_amount=10000000000,        # 10,000 USDC (6 decimals)
    buy_amount=5000000000000000000, # 5 WETH minimum (18 decimals)
    start_type=StartType.AT_MINING_TIME,
    number_of_parts=10,
    time_between_parts=3600,        # 1 hour between each part
    duration_type=DurationType.AUTO,
    app_data="0x" + "0" * 64,
    start_time_epoch=0,
    duration_of_part=0
)

twap_order = Twap.from_data(twap_data)
twap_order.assert_is_valid()

print(f"TWAP Order ID: {twap_order.id}")
print(f"Parts: {twap_data.number_of_parts}")
print(f"Amount per part: {twap_data.sell_amount / twap_data.number_of_parts}")

​
TWAP Parameters

​
Token Configuration

ParameterTypeDescription
sell_tokenstringAddress of the token to sell (checksummed)
buy_tokenstringAddress of the token to buy (checksummed)
sell_amountintegerTotal amount to sell across all parts
buy_amountintegerMinimum total amount to receive across all parts

​
Timing Configuration

ParameterTypeDescription
start_typeStartTypeAT_MINING_TIME or AT_EPOCH
number_of_partsintegerNumber of parts (must be greater than 1 and at most 2^32)
time_between_partsintegerSeconds between each part (1 to 31,536,000)
duration_typeDurationTypeAUTO or LIMIT_DURATION

​
Start Types

​
AT_MINING_TIME

The TWAP starts when the order creation transaction is mined: 
twap_data = TwapData(
    start_type=StartType.AT_MINING_TIME,
    start_time_epoch=0,  # Not used
    # ... other parameters
)

​
AT_EPOCH

Schedule the TWAP to start at a specific time: 
import time

start_time = int(time.time()) + 86400  # Start 24 hours from now

twap_data = TwapData(
    start_type=StartType.AT_EPOCH,
    start_time_epoch=start_time,
    # ... other parameters
)

​
Duration Types

​
AUTO Duration

Each part is valid until the next part starts: 
twap_data = TwapData(
    duration_type=DurationType.AUTO,
    duration_of_part=0,        # Not used
    time_between_parts=3600,   # Each part valid for 1 hour
    # ... other parameters
)
# Part 1: Valid from hour 0 to hour 1
# Part 2: Valid from hour 1 to hour 2

​
LIMIT_DURATION

Set a custom validity period for each part: 
twap_data = TwapData(
    duration_type=DurationType.LIMIT_DURATION,
    time_between_parts=3600,   # Parts start 1 hour apart
    duration_of_part=1800,     # Each part valid for only 30 minutes
    # ... other parameters
)
The duration_of_part must be less than or equal to time_between_parts or the order will be invalid.

​
Submitting a TWAP Order

from web3 import Web3, Account

twap_order = Twap.from_data(twap_data)
create_calldata = twap_order.create_calldata

composable_cow_address = Web3.to_checksum_address("0xfdaFc9d1902f4e0b84f65F49f244b32b31013b74")
provider = Web3(Web3.HTTPProvider("https://rpc.ankr.com/eth"))
account = Account.from_key("YOUR_PRIVATE_KEY")

tx = {
    'from': account.address,
    'to': composable_cow_address,
    'data': create_calldata,
    'gas': 500000,
    'gasPrice': provider.eth.gas_price,
    'nonce': provider.eth.get_transaction_count(account.address),
    'chainId': 1
}

signed_tx = account.sign_transaction(tx)
tx_hash = provider.eth.send_raw_transaction(signed_tx.rawTransaction)
print(f"Transaction hash: {tx_hash.hex()}")

​
Monitoring TWAP Execution

import asyncio
from cowdao_cowpy.composable.types import PollParams, PollResultCode
from cowdao_cowpy.order_book.api import OrderBookApi
from web3 import AsyncWeb3

async def monitor_twap(twap_order, owner_address, chain):
    provider = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider("https://rpc.ankr.com/eth"))
    order_book_api = OrderBookApi()

    poll_params = PollParams(
        owner=owner_address,
        chain=chain,
        provider=provider,
        order_book_api=order_book_api
    )

    while True:
        result = await twap_order.poll(poll_params)

        if result.result == PollResultCode.SUCCESS:
            print(f"Part ready to execute!")
            print(f"Order: {result.order}")
            break
        elif result.result == PollResultCode.TRY_AT_EPOCH:
            next_time = result.epoch
            print(f"Next part starts at epoch {next_time}")
            await asyncio.sleep(max(0, next_time - int(time.time())))
        elif result.result == PollResultCode.DONT_TRY_AGAIN:
            print(f"TWAP completed or invalid: {result.reason}")
            break
        else:
            print(f"Waiting: {result.reason}")
            await asyncio.sleep(60)

​
TWAP Status and Validation

from cowdao_cowpy.composable.types import OwnerParams
from web3 import AsyncWeb3

async def check_twap_status(twap_order, owner_address, chain):
    provider = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider("https://rpc.ankr.com/eth"))

    params = OwnerParams(owner=owner_address, chain=chain, provider=provider)

    is_authorized = await twap_order.is_authorized(params)
    print(f"Authorized: {is_authorized}")

    poll_params = PollParams(
        owner=owner_address, chain=chain,
        provider=provider, order_book_api=OrderBookApi()
    )

    start_ts = await twap_order.start_timestamp(poll_params)
    end_ts = twap_order.end_timestamp(start_ts)

    print(f"Start time: {start_ts}")
    print(f"End time: {end_ts}")
    print(f"Duration: {end_ts - start_ts} seconds")

    validation = twap_order.is_valid()
    if validation.is_valid:
        print("Order is valid")
    else:
        print(f"Order invalid: {validation.reason}")

​
Canceling a TWAP

remove_calldata = twap_order.remove_calldata

tx = {
    'from': account.address,
    'to': composable_cow_address,
    'data': remove_calldata,
    'gas': 200000,
    'gasPrice': provider.eth.gas_price,
    'nonce': provider.eth.get_transaction_count(account.address),
    'chainId': 1
}

signed_tx = account.sign_transaction(tx)
tx_hash = provider.eth.send_raw_transaction(signed_tx.rawTransaction)
print(f"TWAP removed: {tx_hash.hex()}")

​
Serialization

# Serialize
serialized = twap_order.serialize()

# Deserialize
restored_twap = Twap.deserialize(serialized)
assert twap_order.id == restored_twap.id

# Human-readable string
print(twap_order.to_string())

​
Complete Example

import asyncio
import os
from web3 import Web3, Account, AsyncWeb3
from cowdao_cowpy.composable.order_types.twap import Twap, TwapData, StartType, DurationType
from cowdao_cowpy.composable.types import PollParams
from cowdao_cowpy.order_book.api import OrderBookApi
from cowdao_cowpy.common.chains import Chain

async def main():
    USDC = Web3.to_checksum_address("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")
    WETH = Web3.to_checksum_address("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
    account = Account.from_key(os.getenv("PRIVATE_KEY"))

    twap_data = TwapData(
        sell_token=USDC,
        buy_token=WETH,
        receiver=account.address,
        sell_amount=10000 * 10**6,
        buy_amount=4 * 10**18,
        start_type=StartType.AT_MINING_TIME,
        number_of_parts=10,
        time_between_parts=3600,
        duration_type=DurationType.AUTO,
        app_data="0x" + "0" * 64
    )

    twap = Twap.from_data(twap_data)
    twap.assert_is_valid()

    print(f"Created TWAP Order: {twap.id}")
    print(f"Each part sells: {twap_data.sell_amount / twap_data.number_of_parts / 10**6} USDC")

    provider = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider("https://rpc.ankr.com/eth"))

    poll_params = PollParams(
        owner=account.address,
        chain=Chain.MAINNET,
        provider=provider,
        order_book_api=OrderBookApi()
    )

    result = await twap.poll(poll_params)
    print(f"Poll result: {result.result}")

if __name__ == "__main__":
    asyncio.run(main())

​
Best Practices

  • Choose Appropriate Part Counts: 5-20 parts is typically optimal for mainnet. More parts = better averaging but higher gas.
  • Set Realistic Price Limits: The buy_amount is divided across all parts. Ensure each part’s limit price is achievable.
  • Consider Market Hours: Schedule TWAPs during liquid trading periods.
  • Monitor Cabinet Storage: For AT_MINING_TIME orders, check authorization before polling.
  • Test on Testnets: Always test TWAP strategies on Sepolia first.
Last modified on March 12, 2026