TWAP Orders

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

Parameter	Type	Description
`sell_token`	string	Address of the token to sell (checksummed)
`buy_token`	string	Address of the token to buy (checksummed)
`sell_amount`	integer	Total amount to sell across all parts
`buy_amount`	integer	Minimum total amount to receive across all parts

Timing Configuration

Parameter	Type	Description
`start_type`	StartType	`AT_MINING_TIME` or `AT_EPOCH`
`number_of_parts`	integer	Number of parts (must be greater than 1 and at most 2^32)
`time_between_parts`	integer	Seconds between each part (1 to 31,536,000)
`duration_type`	DurationType	`AUTO` 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.

Composable Orders — Learn about the ConditionalOrder framework
Multi-Chain Trading — Execute TWAPs across different chains
Managing Orders — Create and manage orders via API

Getting Started

Concepts

Guides

Advanced

API Reference

Overview

Creating a TWAP Order

TWAP Parameters

Token Configuration

Timing Configuration

Start Types

AT_MINING_TIME

AT_EPOCH

Duration Types

AUTO Duration

LIMIT_DURATION

Submitting a TWAP Order

Monitoring TWAP Execution

TWAP Status and Validation

Canceling a TWAP

Serialization

Complete Example

Best Practices

​Overview

​Creating a TWAP Order

​TWAP Parameters

​Token Configuration

​Timing Configuration

​Start Types

​AT_MINING_TIME

​AT_EPOCH

​Duration Types

​AUTO Duration

​LIMIT_DURATION

​Submitting a TWAP Order

​Monitoring TWAP Execution

​TWAP Status and Validation

​Canceling a TWAP

​Serialization

​Complete Example

​Best Practices

​Related Resources

Overview

Creating a TWAP Order

TWAP Parameters

Token Configuration

Timing Configuration

Start Types

AT_MINING_TIME

AT_EPOCH

Duration Types

AUTO Duration

LIMIT_DURATION

Submitting a TWAP Order

Monitoring TWAP Execution

TWAP Status and Validation

Canceling a TWAP

Serialization

Complete Example

Best Practices

Related Resources